potential 2.1.0

//! Borrowing for futures - a bit like `Option`, but async and with a return channel.
//! A bit like a `Mutex`, compensating for `MutexGuard` not being safe to send.
//! The item is moved (leased) from the owner Potential into the Lease.
//! When the Lease is dropped, the item is sent back to the owner.
//! Owner of the Potential can await the return of the leased item.
//!
//! It is geared towards asynchronous servers. The use case is that
//! some resources can or should be reused, such as TCP connections
//! or crypto configuration or backend storage but they need to be
//! sent off with futures where there is no track of the _future_
//! of the shared resource unless &mut is used so single copy is
//! guaranteed, Mutex is used with a lifetime bound for the future
//! or you use Potential which allows you not to bind the future lifetime
//! to the source. Still, only one running future/task can use the resource
//! at a time. The owner can thus control the concurrency which is by default 1
//! and can be increased by resetting the Potential to a new value.
//!
//! If the lease is not dropped properly (such as if you `mem::forget()` it
//! or when its thread panics badly), the `Potential` calls will be stuck forever.
//! If you suspect that this may happen, await with a timeout to be able to recover.
//!
//! # Sync + Send future example
//!   Notice the mutation of String through immutable reference,
//!   initializing the potential through a lease,
//!   and that futures are Sync + Send. 
//!   Futures on a lease naturally serialize (are mutually exclusive).
//! ```
//! use potential::Potential;
//! use std::future::Future;
//! use std::task::Poll;
//! use std::pin::Pin;
//! #[derive(Default)]
//! struct Trial {
//!     inner: Potential<String>
//! }
//! impl Trial {
//!     async fn back_to_the_future(&self, item:String) -> usize {
//!         let mut lease = match self.inner.lease().await {
//!             Ok(lease) => lease,
//!             Err(gone) => gone.set(String::new())
//!         };
//!         // here some other async work...
//!         lease.extend(item.chars());
//!         lease.len()
//!     }
//! }
//! async fn test(trial: &Trial) -> String {
//!     let fut2 = trial.back_to_the_future(" and then".to_owned());
//!     let fut1 = trial.back_to_the_future("now".to_owned());
//!     assert_eq!(fut1.await, 3);
//!     assert_eq!(fut2.await, 12);
//!     trial.inner.lease().await.expect("set").clone()
//! }
//! let trial = Trial::default();
//! let mut fut = test(&trial);
//! assert_eq!(Box::pin(fut).as_mut().poll(&mut cx()), Poll::Ready("now and then".to_owned()));
//!
//! is_sync_and_send(test(&trial));
//! fn is_sync_and_send<T: Sync + Send>(_: T) {};
//! fn cx() -> std::task::Context<'static> {
//!     let waker = futures_util::task::noop_waker_ref();
//!     let cx = std::task::Context::from_waker(waker);
//!     cx
//! };
//! ```
//!
//! # Example: 'static + Sync + Send future with Arc
//! Not sure how to pull this off with async fn sugar. Suggestions welcome. 
//! Self must not be captured by the future for this to work.
//! ```
//! use potential::Potential;
//! use std::sync::Arc;
//! use std::future::Future;
//! #[derive(Default)]
//! struct Trial {
//!     inner: Arc<Potential<String>>
//! }
//! impl Trial {
//!     fn back_to_the_static_future(&self, item:String) -> Box<dyn Future<Output = usize> + 'static + Send + Sync> {
//!         let inner = self.inner.clone();
//!         let fut = async move {
//!             let mut lease = match inner.lease_on_arc().await {
//!                 Ok(lease) => lease,
//!                 Err(gone) => gone.set(String::new())
//!             };
//!             // here some other async work...
//!             lease.extend(item.chars());
//!             lease.len()
//!         };
//!         Box::new(fut)
//!     }
//! }
//! let trial = Trial::default();
//! let fut = trial.back_to_the_static_future("now".to_owned());
//! is_sync_and_send_and_static(fut);
//! fn is_sync_and_send_and_static<T: Sync + Send + 'static>(_: T) {};
//! ```

use futures_channel::oneshot::{channel as oneshot, Receiver, Sender};
use futures_util::lock::Mutex;
use log::debug;
use std::fmt;
use std::ops::{Deref, DerefMut};
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

/// The owner of an item leases it out or can access mutably if not leased.
#[derive(Default, Debug)]
pub struct Potential<T> {
    /// Potential state gated with a mutext to allow &ref access
    state: Mutex<State<T>>,
    /// Writes to this field must be gated by successful lock on state, or &mut call.
    /// It simplifies checking the potential state - is_leased/is_present
    leased: AtomicBool,
}
/// Represents the state of the potential within a mutex
#[derive(Debug)]
enum State<T> {
    /// The item is not leased and is with the owner
    Present(T),
    /// The item is currently leased, waiting for a return
    Leased(Receiver<T>),
    /// Transitional state to enable `mem::replace()` or state after stolen lease
    Gone,
}
impl<T> Default for State<T> {
    fn default() -> Self {
        Self::Gone
    }
}
impl<T> Potential<T> {
    /// Create new full Potential
    pub fn new(item: T) -> Self {
        let mut me = Potential::empty();
        me.set(item);
        me
    }
    /// Create new empty Potential
    pub fn empty() -> Self {
        Potential {
            state: Mutex::new(State::Gone),
            leased: AtomicBool::new(false),
        }
    }
    /// Check if the item is waiting for return of the lease right now
    pub fn is_leased(&self) -> bool {
        self.leased.load(Ordering::Relaxed)
    }
    /// Set the current item immediately regardles of lease.
    /// Sucessful lease will not return their item on drop.
    /// Pending and subsequent leases will pick the new value.
    /// ```rust
    /// # use potential::Potential;
    /// let mut potential = Potential::new("x");
    /// potential.set("y");
    /// ```
    /// To set through an immutable reference, feed it back through a lease. Call:
    /// ```rust
    /// # use potential::Potential;
    /// # let potential=Potential::new("x");
    /// # let value = "y";
    /// # let fut = async move {
    /// potential.lease().await?.replace(value);
    /// # Potential::empty().lease().await
    /// # };
    /// ```
    pub fn set(&mut self, item: T) {
        self.state = Mutex::new(State::Present(item));
        self.leased.store(false, Ordering::Relaxed);
    }
    /// Sync + Send + 'static flavor of `lease()` if `Self` is in `Arc`
    pub async fn lease_on_arc(self: Arc<Self>) -> Result<Lease<T>, Gone<T>> {
        self.lease().await
    }
    /// 'static flavor of `lease()` if `Self` is in `Rc`
    pub async fn lease_on_rc(self: Rc<Self>) -> Result<Lease<T>, Gone<T>> {
        self.lease().await
    }
    /// Wait for the item to be available and then access the mutable reference if available.
    /// This call will return None if previous Lease failed to return the item or if created `empty()`.
    /// If that happens, set a new potential.
    ///
    /// Example:
    /// ```rust
    /// # use potential::Potential;
    /// let mut potential = Potential::new(String::new());
    /// # let fut = async move {
    /// potential.get_mut().await?.push('A');
    /// # Some(())
    /// # };
    /// ```
    pub async fn get_mut(&mut self) -> Option<&mut T> {
        let state = self.state.get_mut();
        loop {
            break match state {
                State::Gone => None,
                State::Present(present) => {
                    self.leased.store(false, Ordering::Relaxed);
                    Some(present)
                }
                State::Leased(receiver) => {
                    *state = Self::await_return(receiver).await;
                    self.leased.store(false, Ordering::Relaxed);
                    continue;
                }
            };
        }
    }
    /// Wait for the return of the item and if available, lease it.
    /// This call will return Err(Gone) if previous Lease failed to return the item or if created `empty()`.
    /// If that happens, set a new potential on the Gone.
    ///
    /// Example:
    /// ```rust
    /// # use potential::Potential;
    /// let mut potential = Potential::new(String::new());
    /// # let fut = async move {
    /// potential.lease().await?.push('A');
    /// # Potential::empty().lease().await
    /// # };
    /// ```
    ///
    /// Example of empty potential:
    /// ```rust
    /// # use potential::Potential;
    /// let mut potential = Potential::empty();
    /// # let fut = async move {
    /// let mut lease = match potential.lease().await {
    ///     Err(gone) => gone.set(String::new()),
    ///     Ok(lease) => lease
    /// };
    /// lease.push('A');
    /// # };
    /// ```
    pub async fn lease(&self) -> Result<Lease<T>, Gone<T>> {
        let mut state = self.state.lock().await;
        loop {
            break match std::mem::replace(state.deref_mut(), State::Gone) {
                State::Gone => {
                    let (sender, receiver) = oneshot();
                    *state = State::Leased(receiver);
                    self.leased.store(true, Ordering::Relaxed);
                    Err(Gone::new(sender))
                }
                State::Present(item) => {
                    let (sender, receiver) = oneshot();
                    *state = State::Leased(receiver);
                    self.leased.store(true, Ordering::Relaxed);
                    Ok(Lease::new(item, sender))
                }
                State::Leased(mut receiver) => {
                    *state = Self::await_return(&mut receiver).await;
                    self.leased.store(false, Ordering::Relaxed);
                    continue;
                }
            };
        }
    }
    /// Shared handling of item return
    async fn await_return(receiver: &mut Receiver<T>) -> State<T> {
        if let Ok(item) = receiver.await {
            State::Present(item)
        } else {
            // This could perhaps happen if the thread having the lease panicked badly
            // or after `mem::forget(lease)`
            debug!("Lease dropped without sending the item back. Subsequent call will panic unless you set a new potential.");
            State::Gone
        }
    }
}

/// The leased item. Item will be sent back to the owner on `drop()` unless stolen with `steal()`.
///
/// Lease does not allow empty state so as long as it exists, it must have the item.
///
/// As long as `Lease` exists, `Potential` will be waiting for return (unless reset). Drop `Lease` or steal it to release the wait.
#[derive(Debug)]
pub struct Lease<T> {
    /// The item wrapped in option only to enable `drop()` and `steal()`
    item: Option<T>,
    /// The item owner wrapped in option only to enable `drop()` and `steal()`
    owner: Option<Sender<T>>,
}

impl<T> Lease<T> {
    fn new(item: T, owner: Sender<T>) -> Self {
        Lease {
            item: Some(item),
            owner: Some(owner),
        }
    }
    /// Replace the item within the lease
    pub fn replace(&mut self, replacement: T) -> T {
        std::mem::replace(&mut self.item, Some(replacement)).expect("item must be set")
    }
    /// Steal the leased item destroying the lease.
    /// `Potential` await calls will return `None`.
    pub fn steal(mut self) -> T {
        let item = self.item.take().expect("item must be set");
        let owner = self.owner.take().expect("owner must be set");
        drop(owner);
        std::mem::forget(self);
        item
    }
}
impl<T> Deref for Lease<T> {
    type Target = T;
    fn deref(&self) -> &Self::Target {
        self.item.as_ref().expect("item must be set")
    }
}
impl<T> DerefMut for Lease<T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        self.item.as_mut().expect("item must be set")
    }
}
impl<T> Drop for Lease<T> {
    /// `drop()` must be called for the item to return to the owner.
    /// Otherwise, the owner will be stuck waiting forever.
    fn drop(&mut self) {
        if let Some(owner) = self.owner.take() {
            if let Some(item) = self.item.take() {
                // if there is nobody listening, that's OK
                drop(owner.send(item));
            }
        }
    }
}

/// Error result of a lease on an emtpy `Potential` which can be upgraded to a `Lease` by setting the item.
///
/// As long as `Gone` exists, `Potential` will be waiting for return (unless reset). Drop `Gone` to release the wait.
#[derive(Debug)]
pub struct Gone<T> {
    /// The item owner wrapped in option only to enable `drop()` and `steal()`
    owner: Sender<T>,
}
impl<T> Gone<T> {
    fn new(owner: Sender<T>) -> Self {
        Gone { owner }
    }
    /// Set the item and upgrade to a `Lease`
    pub fn set(self, item: T) -> Lease<T> {
        Lease::new(item, self.owner)
    }
}
impl<T> fmt::Display for Gone<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("Potential item is gone")
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::future::Future;
    use std::task::{Context, Poll};

    fn is_sync<T: Sync>(_: T) {}
    fn is_send<T: Send>(_: T) {}
    fn is_static<T: 'static>(_: T) {}

    #[test]
    fn lease_is_sync() {
        let sut = Potential::new(true);
        is_sync(sut.lease());
    }
    #[test]
    fn lease_is_send() {
        let sut = Potential::new(true);
        is_send(sut.lease());
    }
    #[test]
    fn lease_on_rc_is_static() {
        let sut = Rc::new(Potential::new(true));
        is_static(sut.lease_on_rc());
    }
    #[test]
    fn lease_on_arc_is_static() {
        let sut = Arc::new(Potential::new(true));
        is_static(sut.lease_on_arc());
    }
    #[test]
    fn lease_on_arc_is_sync() {
        let sut = Arc::new(Potential::new(true));
        is_sync(sut.lease_on_arc());
    }
    #[test]
    fn lease_on_arc_is_send() {
        let sut = Arc::new(Potential::new(true));
        is_send(sut.lease_on_arc());
    }

    #[test]
    fn it_blocks() {
        let sut = Potential::new(true);
        let mut lease1_fut = Box::pin(sut.lease());
        let mut lease2_fut = Box::pin(sut.lease());
        // first lease successfully received
        let lease1 = ready_ok(lease1_fut.as_mut().poll(&mut cx()));
        // lease2 cannot proceed because the value is in lease1
        assert!(
            lease2_fut.as_mut().poll(&mut cx()).is_pending(),
            "Second lease is blocked by the first"
        );
        // after dropping lease1, lease2 will go ahead.
        drop(lease1);
        let lease2 = ready_ok(lease2_fut.as_mut().poll(&mut cx()));
        drop(lease2);
    }

    #[test]
    fn it_hangs_on_no_drop() {
        let sut = Potential::new(true);
        let mut lease1_fut = Box::pin(sut.lease());
        let mut lease2_fut = Box::pin(sut.lease());
        // first lease successfully received
        let lease1 = ready_ok(lease1_fut.as_mut().poll(&mut cx()));
        // lease2 cannot proceed because the value is in lease1
        assert!(
            lease2_fut.as_mut().poll(&mut cx()).is_pending(),
            "Second lease is blocked by the first"
        );
        // after forgetting lease1 (not calling a destructor), lease2 will be stuck forever.
        std::mem::forget(lease1);
        assert!(
            lease2_fut.as_mut().poll(&mut cx()).is_pending(),
            "Second lease is blocked by the first"
        );
    }
    #[test]
    fn it_handles_stealing() {
        let sut = Potential::new(true);
        let mut lease1_fut = Box::pin(sut.lease());
        let mut lease2_fut = Box::pin(sut.lease());
        // first lease successfully received
        let lease1 = ready_ok(lease1_fut.as_mut().poll(&mut cx()));
        // lease2 cannot proceed because the value is in lease1
        assert!(
            lease2_fut.as_mut().poll(&mut cx()).is_pending(),
            "Second lease is blocked by the first"
        );
        // after stealing lease1 (not calling a destructor), lease2 will return None.
        lease1.steal();
        ready_err(lease2_fut.as_mut().poll(&mut cx()));
    }

    fn ready_ok<T, E>(poll: Poll<Result<T, E>>) -> T {
        match poll {
            Poll::Ready(Ok(t)) => t,
            _ => panic!("poll is not ready or is err"),
        }
    }
    fn ready_err<T, E>(poll: Poll<Result<T, E>>) -> E {
        match poll {
            Poll::Ready(Err(e)) => e,
            _ => panic!("poll is not ready or is ok"),
        }
    }

    fn cx() -> Context<'static> {
        let waker = futures_util::task::noop_waker_ref();
        let cx = Context::from_waker(waker);
        cx
    }
}