#![forbid(unsafe_code)]

//! This crate allows asynchronous value exchanges between two endpoints.
//!
//! Conceptually this is similar to having two channels, each with capacity 1.
//!
//! # Example
//!
//! Create an exchange between a client *A* and a server *B* where *A* sends
//! `u32` values to *B* and *B* acknowledges each value with a `bool`, `true`
//! if the value is odd and `false` otherwise.
//!
//! ```rust
//! # async fn doc() {
//! let (mut a, mut b) = scambio::exchange();
//!
//! let client = async move {
//!     for i in 0 .. 10u32 {
//!         assert!(a.send(i).await.is_ok());
//!         assert_eq!(Some(i % 2 == 1), a.receive().await)
//!     }
//! };
//!
//! let server = async move {
//!     while let Some(i) = b.receive().await {
//!         assert!(b.send(i % 2 == 1).await.is_ok())
//!     }
//! };
//!
//! assert_eq!(futures::join!(client, server), ((), ()));
//! # }
//! ```

use parking_lot::Mutex;
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll, Waker};

mod error;
pub use error::Error;

/// An exchange endpoint for value transfers.
///
/// Every exchange consists of two halves, [`Left`] and [`Right`], which can
/// transfer data between each other one by one. At most one data item is
/// stored per endpoint half.
pub struct Left<L, R> {
    data: Arc<Mutex<SharedData<L, R>>>
}

/// An exchange endpoint for value transfers.
///
/// Every exchange consists of two halves, [`Left`] and [`Right`], which can
/// transfer data between each other one by one. At most one data item is
/// stored per endpoint half.
pub struct Right<L, R> {
    data: Arc<Mutex<SharedData<L, R>>>
}

/// Shared data between endpoints.
struct SharedData<L, R> {
    /// Has `Left` not been dropped yet?
    l_alive: bool,
    /// Has `Right` not been dropped yet?
    r_alive: bool,
    /// The value `Left` wants to transfer to `Right`.
    l_offer: Option<L>,
    /// The value `Right` wants to transfer to `Left`.
    r_offer: Option<R>,
    /// If `r_offer` is empty, `Left`'s waker is stored in here and
    /// woken up once `Right` has put a value in `r_offer`, enabling
    /// `Left` to get the value.
    l_waker_r: Option<Waker>,
    /// If `l_offer` is full, `Left`'s waker is stored in here and
    /// woken up once `Right` has accepted `l_offer`, enabling `Left` to
    /// continue.
    l_waker_w: Option<Waker>,
    /// If `l_offer` is empty, `Right`'s waker is stored in here and
    /// woken up once `Left` has put a value in `l_offer`, enabling
    /// `Right` to get the value.
    r_waker_r: Option<Waker>,
    /// If `r_offer` is full, `Right`'s waker is stored in here and
    /// woken up once `Left` has accepted `r_offer`, enabling `Right` to
    /// continue.
    r_waker_w: Option<Waker>
}

/// Create a new exchange between `Left` and `Right`.
pub fn exchange<L, R>() -> (Left<L, R>, Right<L, R>) {
    let d = SharedData {
        l_alive: true,
        r_alive: true,
        l_offer: None,
        r_offer: None,
        l_waker_r: None,
        l_waker_w: None,
        r_waker_r: None,
        r_waker_w: None
    };
    let l = Left {
        data: Arc::new(Mutex::new(d))
    };
    let r = Right {
        data: l.data.clone()
    };
    (l, r)
}

impl<L, R> Left<L, R> {
    /// Has the other end been dropped?
    pub fn is_closed(&self) -> bool {
        !self.data.lock().r_alive
    }

    /// Send a value to [`Right`].
    ///
    /// If `Right` has been dropped, the value will be returned.
    pub async fn send(&mut self, value: L) -> Result<(), L> {
        SendReadyL(self).await;
        self.send_now(value).map_err(Error::into_data)
    }

    /// Receive a value from [`Right`].
    ///
    /// If `Right` has been dropped, `None` will be returned.
    pub async fn receive(&mut self) -> Option<R> {
        ReceiveR2L(self).await
    }

    /// Receive a value from [`Right`] if available.
    ///
    /// If `Right` has been dropped, `Poll::Ready(None)` will be returned.
    /// If no value is available, `Poll::Pending` will be returned and `Left` will
    /// be woken up once a value can be received for `Right`.
    pub fn poll_receive(&mut self, cx: &mut Context) -> Poll<Option<R>> {
        let mut data = self.data.lock();
        if let Some(val) = data.r_offer.take() {
            if let Some(w) = data.r_waker_w.take() {
                w.wake()
            }
            return Poll::Ready(Some(val))
        }
        if !data.r_alive {
            return Poll::Ready(None)
        }
        if !data.l_waker_r.as_ref().map_or(false, |w| w.will_wake(cx.waker())) {
            data.l_waker_r = Some(cx.waker().clone())
        }
        Poll::Pending
    }

    /// Check if there is capacity to send a data item to [`Right`].
    pub fn poll_send_ready(&mut self, cx: &mut Context) -> Poll<()> {
        let mut data = self.data.lock();
        if data.l_offer.is_none() {
            return Poll::Ready(())
        }
        if !data.l_waker_w.as_ref().map_or(false, |w| w.will_wake(cx.waker())) {
            data.l_waker_w = Some(cx.waker().clone())
        }
        Poll::Pending
    }

    /// Try to send data to [`Right`].
    ///
    /// If `Right` has been dropped or there is already a pending data item for
    /// `Right` the data will be returned to the caller.
    ///
    /// Use [`Left::poll_send_ready`] to ensure there is capacity available.
    pub fn send_now(&mut self, value: L) -> Result<(), Error<L>> {
        let mut data = self.data.lock();
        if !data.r_alive {
            return Err(Error::Closed(value))
        }
        if data.l_offer.is_some() {
            return Err(Error::AtCapacity(value))
        }
        data.l_offer = Some(value);
        if let Some(w) = data.r_waker_r.take() {
            w.wake()
        }
        Ok(())
    }
}

impl<L, R> Right<L, R> {
    /// Has the other end been dropped?
    pub fn is_closed(&self) -> bool {
        !self.data.lock().l_alive
    }

    /// Send a value to [`Left`].
    ///
    /// If `Left` has been dropped, the value will be returned.
    pub async fn send(&mut self, value: R) -> Result<(), R> {
        SendReadyR(self).await;
        self.send_now(value).map_err(Error::into_data)
    }

    /// Receive a value from [`Left`].
    ///
    /// If `Left` has been dropped, `None` will be returned.
    pub async fn receive(&mut self) -> Option<L> {
        ReceiveL2R(self).await
    }

    /// Receive a value from [`Left`] if available.
    ///
    /// If `Left` has been dropped, `Poll::Ready(None)` will be returned.
    /// If no value is available, `Poll::Pending` will be returned and `Right` will
    /// be woken up once a value can be received for `Left`.
    pub fn poll_receive(&mut self, cx: &mut Context) -> Poll<Option<L>> {
        let mut data = self.data.lock();
        if let Some(val) = data.l_offer.take() {
            if let Some(w) = data.l_waker_w.take() {
                w.wake()
            }
            return Poll::Ready(Some(val))
        }
        if !data.l_alive {
            return Poll::Ready(None)
        }
        if !data.r_waker_r.as_ref().map_or(false, |w| w.will_wake(cx.waker())) {
            data.r_waker_r = Some(cx.waker().clone())
        }
        Poll::Pending
    }

    /// Check if there is capacity to send a data item to [`Left`].
    pub fn poll_send_ready(&mut self, cx: &mut Context) -> Poll<()> {
        let mut data = self.data.lock();
        if data.r_offer.is_none() {
            return Poll::Ready(())
        }
        if !data.r_waker_w.as_ref().map_or(false, |w| w.will_wake(cx.waker())) {
            data.r_waker_w = Some(cx.waker().clone())
        }
        Poll::Pending
    }

    /// Try to send data to [`Left`].
    ///
    /// If `Left` has been dropped or there is already a pending data item for
    /// `Left` the data will be returned to the caller.
    ///
    /// Use [`Right::poll_send_ready`] to ensure there is capacity available.
    pub fn send_now(&mut self, value: R) -> Result<(), Error<R>> {
        let mut data = self.data.lock();
        if !data.l_alive {
            return Err(Error::Closed(value))
        }
        if data.r_offer.is_some() {
            return Err(Error::AtCapacity(value))
        }
        data.r_offer = Some(value);
        if let Some(w) = data.l_waker_r.take() {
            w.wake()
        }
        Ok(())
    }
}

impl<L, R> Unpin for Left<L, R> {}
impl<L, R> Unpin for Right<L, R> {}

impl<L, R> fmt::Debug for Left<L, R> {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("Left")
    }
}

impl<L, R> fmt::Debug for Right<L, R> {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("Right")
    }
}

impl<L, R> Drop for Left<L, R> {
    fn drop(&mut self) {
        let mut data = self.data.lock();
        data.l_alive = false;
        if let Some(w) = data.r_waker_r.take() {
            w.wake()
        }
        if let Some(w) = data.r_waker_w.take() {
            w.wake()
        }
    }
}

impl<L, R> Drop for Right<L, R> {
    fn drop(&mut self) {
        let mut data = self.data.lock();
        data.r_alive = false;
        if let Some(w) = data.l_waker_r.take() {
            w.wake()
        }
        if let Some(w) = data.l_waker_w.take() {
            w.wake()
        }
    }
}

/// A future receiving a value from `Right`.
struct ReceiveR2L<'a, L, R>(&'a mut Left<L, R>);

/// A future receiving a value from `Left`.
struct ReceiveL2R<'b, L, R>(&'b mut Right<L, R>);

/// A future checking if there is capacity for `Left` to send data.
struct SendReadyL<'a, L, R>(&'a mut Left<L, R>);

/// A future checking if there is capacity for `Right` to send data.
struct SendReadyR<'b, L, R>(&'b mut Right<L, R>);

impl<L, R> Future for ReceiveR2L<'_, L, R> {
    type Output = Option<R>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
        self.0.poll_receive(cx)
    }
}

impl<L, R> Future for ReceiveL2R<'_, L, R> {
    type Output = Option<L>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
        self.0.poll_receive(cx)
    }
}

impl<L, R> Future for SendReadyL<'_, L, R> {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
        self.0.poll_send_ready(cx)
    }
}

impl<L, R> Future for SendReadyR<'_, L, R> {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
        self.0.poll_send_ready(cx)
    }
}