ufotofu 0.10.1

//! A **S**ingle-**S**ender **S**ingle-**R**eceiver FIFO channel.
//!
//! The entrypoint to this module is the [`new_sssr`] function. You supply it with a [`Queue`], and it returns the two endpoints of the channel: a [`Sender`] and a [`Receiver`].
//!
//! Calling [`consume`](Consumer::consume) with a `Left` on the [`Sender`] results in the item being emitted on the [`Receiver`]. If the internal queue is full, the `consume` future remains pending until the [`Receiver`] has produced at least one item.
//!
//! Calling [`consume`](Consumer::consume) with a `Right` on the [`Sender`] results in the final value being emitted on the [`Receiver`] (though all previously sent items are delivered first).
//!
//!  Dropping the [`Sender`] or the [`Receiver`] does not actively notify the other endpoint about anything. But both endpoint kinds have synchronous methods for checking whether there other endpoint is still active.
//!
//! #### Example
//!
//! ```
//! use futures::join;
//! use ufotofu::prelude::*;
//! use ufotofu::channels::sssr::*;
//!
//! // Create the channel, using a queue of capacity two.
//! let (mut sender, mut receiver) = new_sssr(ufotofu::queues::new_fixed::<i16>(2));
//!
//! pollster::block_on(async {
//!     // A future sending three items to the channel, then closing.
//!     let send_things = async {
//!         assert!(sender.consume_item(300).await.is_ok());
//!         assert!(sender.consume_item(400).await.is_ok());
//!         assert!(sender.consume_item(500).await.is_ok());
//!         assert!(sender.consume_final(-17).await.is_ok());
//!     };
//!
//!     // A future receiving the items from the channel.
//!     let receive_things = async {
//!         assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
//!         assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
//!         assert_eq!(500, receiver.produce().await.unwrap().unwrap_left());
//!         assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
//!     };
//!
//!     // Concurrently send and receive the items. Concurrency is necessary, because
//!     // the number of transmitted items exceeds the maximum capacity of the queue we use.
//!     join!(receive_things, send_things);
//! });
//! ```

use core::{convert::Infallible, fmt};

use std::rc::Rc;

use either::Either::{self};

use crate::prelude::*;
use crate::queues::Queue;

use super::advanced::sssr::{Receiver as AdvancedReceiver, Sender as AdvancedSender, State};

/// Creates a new single-sender single-receiver channel in the form of a [`Sender`] and a [`Receiver`] endpoint, which communicate via the given [`Queue`].
///
/// #### Example
///
/// ```
/// use futures::join;
/// use ufotofu::prelude::*;
/// use ufotofu::channels::sssr::*;
///
/// // Create the channel, using a queue of capacity two.
/// let (mut sender, mut receiver) = new_sssr(ufotofu::queues::new_fixed::<i16>(2));
///
/// pollster::block_on(async {
///     // A future sending three items to the channel, then closing.
///     let send_things = async {
///         assert!(sender.consume_item(300).await.is_ok());
///         assert!(sender.consume_item(400).await.is_ok());
///         assert!(sender.consume_item(500).await.is_ok());
///         assert!(sender.consume_final(-17).await.is_ok());
///     };
///
///     // A future receiving the items from the channel.
///     let receive_things = async {
///         assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
///         assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
///         assert_eq!(500, receiver.produce().await.unwrap().unwrap_left());
///         assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
///     };
///
///     // Concurrently send and receive the items. Concurrency is necessary, because
///     // the number of transmitted items exceeds the maximum capacity of the queue we use.
///     join!(receive_things, send_things);
/// });
/// ```
pub fn new_sssr<Q: Queue, F>(queue: Q) -> (Sender<Q, F>, Receiver<Q, F>) where {
    let state = Rc::new(State::new(queue));
    let (s, r) = super::advanced::new_sssr(state);
    (Sender(s), Receiver(r))
}

/// An endpoint for sending items to an sssr channel (and for closing the channel).
///
/// Use the [`Consumer`] or [`BulkConsumer`] implementations to write data to the channel.
///
/// This type does *not* implement [`Clone`], making the channel a *single-sender* channel.
///
/// <br/>Counterpart: the [`Receiver`] type.
pub struct Sender<Q, F>(AdvancedSender<Rc<State<Q, F>>, Q, F>);

impl<Q, F> fmt::Debug for Sender<Q, F>
where
    Q: fmt::Debug,
    F: fmt::Debug,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(f)
    }
}

impl<Q, F> Sender<Q, F>
where
    Q: Queue,
{
    /// Returns the number of items in the queue that buffers items for this channel.
    ///
    /// Note that this method only considers regular items — whether the final value has been written to the channel does not influence this method at all.
    ///
    /// ```
    /// use futures::join;
    /// use ufotofu::prelude::*;
    /// use ufotofu::channels::sssr::*;
    ///
    /// let (mut sender, mut receiver) = new_sssr(ufotofu::queues::new_static::<i16, 2>());
    ///
    /// pollster::block_on(async {
    ///     assert_eq!(sender.len(), 0);
    ///     assert!(sender.consume_item(300).await.is_ok());
    ///     assert_eq!(sender.len(), 1);
    ///     assert!(sender.consume_item(400).await.is_ok());
    ///     assert_eq!(sender.len(), 2);
    ///
    ///     assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
    ///     assert_eq!(sender.len(), 1);
    ///
    ///     assert!(sender.consume_final(-17).await.is_ok());
    ///     assert_eq!(sender.len(), 1);
    ///
    ///     assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
    ///     assert_eq!(sender.len(), 0);
    ///     assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
    ///     assert_eq!(sender.len(), 0);
    /// });
    /// ```
    pub fn len(&self) -> usize {
        self.0.len()
    }

    /// Returns whether the queue that buffers items for this channel is currently empty.
    ///
    /// Note that this method only considers regular items — whether the final value has been written to the channel does not influence this method at all.
    ///
    /// ```
    /// use futures::join;
    /// use ufotofu::prelude::*;
    /// use ufotofu::channels::sssr::*;
    ///
    /// let (mut sender, mut receiver) = new_sssr(ufotofu::queues::new_static::<i16, 2>());
    ///
    /// pollster::block_on(async {
    ///     assert!(sender.is_empty());
    ///     assert!(sender.consume_item(300).await.is_ok());
    ///     assert!(!sender.is_empty());
    ///     assert!(sender.consume_item(400).await.is_ok());
    ///     assert!(!sender.is_empty());
    ///
    ///     assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
    ///     assert!(!sender.is_empty());
    ///
    ///     assert!(sender.consume_final(-17).await.is_ok());
    ///     assert!(!sender.is_empty());
    ///
    ///     assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
    ///     assert!(sender.is_empty());
    ///     assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
    ///     assert!(sender.is_empty());
    /// });
    /// ```
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    /// Returns whether the corresponding [`Receiver`] has been dropped already.
    ///
    /// ```
    /// use ufotofu::channels::sssr::*;
    ///
    /// let (sender, receiver) = new_sssr::<_, ()>(ufotofu::queues::new_static::<i16, 2>());
    ///
    /// assert!(!sender.is_receiver_dropped());
    /// core::mem::drop(receiver);
    /// assert!(sender.is_receiver_dropped());
    /// ```
    ///
    /// <br/>Counterpart: the [`Receiver::is_sender_dropped`] method.
    pub fn is_receiver_dropped(&self) -> bool {
        self.0.is_receiver_dropped()
    }
}

impl<Q: Queue, F> Consumer for Sender<Q, F> {
    type Item = Q::Item;
    type Final = F;
    type Error = Infallible;

    /// Writes the item into the buffer queue, waiting for buffer space to
    /// become available (by reading items from the corresponding [`Sender`]) if necessary.
    async fn consume(&mut self, val: Either<Self::Item, Self::Final>) -> Result<(), Self::Error> {
        self.0.consume(val).await
    }

    async fn flush(&mut self) -> Result<(), Self::Error> {
        self.0.flush().await
    }
}

impl<Q: Queue, F> BulkConsumer for Sender<Q, F> {
    async fn expose_slots_gracefully<Fun, Ret>(&mut self, f: Fun) -> Result<Ret, (Fun, Self::Error)>
    where
        Fun: AsyncFnOnce(&mut [Self::Item]) -> (usize, Ret),
    {
        self.0.expose_slots_gracefully(f).await
    }
}

/// An endpoint for receiving items from an sssr channel.
///
/// Use the [`Producer`] or [`BulkProducer`] implementations to read data from the channel.
///
/// This type does *not* implement [`Clone`], making the channel a *single-receiver* channel.
///
/// <br/>Counterpart: the [`Sender`] type.
pub struct Receiver<Q, F>(AdvancedReceiver<Rc<State<Q, F>>, Q, F>);

impl<Q, F> fmt::Debug for Receiver<Q, F>
where
    Q: fmt::Debug,
    F: fmt::Debug,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(f)
    }
}

impl<Q: Queue, F> Receiver<Q, F> {
    /// Returns the number of items in the queue that buffers items for this channel.
    ///
    /// Note that this method only considers regular items — whether the final value has been written to the channel does not influence this method at all.
    ///
    /// ```
    /// use futures::join;
    /// use ufotofu::prelude::*;
    /// use ufotofu::channels::sssr::*;
    ///
    /// let (mut sender, mut receiver) = new_sssr(ufotofu::queues::new_static::<i16, 2>());
    ///
    /// pollster::block_on(async {
    ///     assert_eq!(receiver.len(), 0);
    ///     assert!(sender.consume_item(300).await.is_ok());
    ///     assert_eq!(receiver.len(), 1);
    ///     assert!(sender.consume_item(400).await.is_ok());
    ///     assert_eq!(receiver.len(), 2);
    ///
    ///     assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
    ///     assert_eq!(receiver.len(), 1);
    ///
    ///     assert!(sender.consume_final(-17).await.is_ok());
    ///     assert_eq!(receiver.len(), 1);
    ///
    ///     assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
    ///     assert_eq!(receiver.len(), 0);
    ///     assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
    ///     assert_eq!(receiver.len(), 0);
    /// });
    /// ```
    pub fn len(&self) -> usize {
        self.0.len()
    }

    /// Returns whether the queue that buffers items for this channel is currently empty.
    ///
    /// Calling [`produce`](Producer::produce) on this receiver yields a *pending* future if and only if this returns `true`.
    ///
    /// Note that this method only considers regular items — whether the final value has been written to the channel does not influence this method at all.
    ///
    /// ```
    /// use futures::join;
    /// use ufotofu::prelude::*;
    /// use ufotofu::channels::sssr::*;
    ///
    /// let (mut sender, mut receiver) = new_sssr(ufotofu::queues::new_static::<i16, 2>());
    ///
    /// pollster::block_on(async {
    ///     assert!(receiver.is_empty());
    ///     assert!(sender.consume_item(300).await.is_ok());
    ///     assert!(!receiver.is_empty());
    ///     assert!(sender.consume_item(400).await.is_ok());
    ///     assert!(!receiver.is_empty());
    ///
    ///     assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
    ///     assert!(!receiver.is_empty());
    ///
    ///     assert!(sender.consume_final(-17).await.is_ok());
    ///     assert!(!receiver.is_empty());
    ///
    ///     assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
    ///     assert!(receiver.is_empty());
    ///     assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
    ///     assert!(receiver.is_empty());
    /// });
    /// ```
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    /// Returns whether the corresponding [`Sender`] has been dropped already.
    ///
    /// ```
    /// use ufotofu::channels::sssr::*;
    ///
    /// let (sender, receiver) = new_sssr::<_, ()>(ufotofu::queues::new_static::<i16, 2>());
    ///
    /// assert!(!receiver.is_sender_dropped());
    /// core::mem::drop(sender);
    /// assert!(receiver.is_sender_dropped());
    /// ```
    ///
    /// <br/>Counterpart: the [`Sender::is_receiver_dropped`] method.
    pub fn is_sender_dropped(&self) -> bool {
        self.0.is_sender_dropped()
    }
}

impl<Q: Queue, F> Producer for Receiver<Q, F> {
    type Item = Q::Item;
    type Final = F;
    type Error = Infallible;

    /// Take an item from the buffer queue, waiting for an item to
    /// become available (by being consumed by the corresponding [`Sender`]) if necessary.
    async fn produce(&mut self) -> Result<Either<Self::Item, Self::Final>, Self::Error> {
        self.0.produce().await
    }

    async fn slurp(&mut self) -> Result<(), Self::Error> {
        self.0.slurp().await
    }
}

impl<Q: Queue, F> BulkProducer for Receiver<Q, F> {
    async fn expose_items_gracefully<Fun, Ret>(
        &mut self,
        f: Fun,
    ) -> Result<Either<Ret, (Fun, Self::Final)>, (Fun, Self::Error)>
    where
        Fun: AsyncFnOnce(&[Self::Item]) -> (usize, Ret),
    {
        self.0.expose_items_gracefully(f).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use futures::join;

    use crate::queues::new_static;

    #[test]
    fn test_spsc_sufficient_capacity() {
        let (mut sender, mut receiver) = new_sssr(new_static::<i16, 99>());

        pollster::block_on(async {
            assert!(sender.consume_item(300).await.is_ok());
            assert!(sender.consume_item(400).await.is_ok());
            assert!(sender.consume_item(500).await.is_ok());
            assert!(sender.consume_final(-17).await.is_ok());
            assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
            assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
            assert_eq!(500, receiver.produce().await.unwrap().unwrap_left());
            assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
        });
    }

    #[test]
    fn test_spsc_low_capacity() {
        pollster::block_on(async {
            let (mut sender, mut receiver) = new_sssr(new_static::<i16, 2>());

            let send_things = async {
                assert!(sender.consume_item(300).await.is_ok());
                assert!(sender.consume_item(400).await.is_ok());
                assert!(sender.consume_item(500).await.is_ok());
                assert!(sender.consume_final(-17).await.is_ok());
            };

            let receive_things = async {
                assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
                assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
                assert_eq!(500, receiver.produce().await.unwrap().unwrap_left());
                assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
            };

            join!(send_things, receive_things);
        });
    }

    #[test]
    fn test_spsc_immediate_final() {
        pollster::block_on(async {
            let (mut sender, mut receiver) = new_sssr(new_static::<i16, 3>());

            let send_things = async {
                assert!(sender.consume_final(-17).await.is_ok());
            };

            let receive_things = async {
                assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
            };

            join!(send_things, receive_things);
        });
    }

    #[test]
    fn test_spsc_receive_then_send_concurrently() {
        pollster::block_on(async {
            let (mut sender, mut receiver) = new_sssr(new_static::<i16, 2>());

            let send_things = async {
                assert!(sender.consume_item(300).await.is_ok());
                assert!(sender.consume_item(400).await.is_ok());
                assert!(sender.consume_item(500).await.is_ok());
                assert!(sender.consume_final(-17).await.is_ok());
            };

            let receive_things = async {
                assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
                assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
                assert_eq!(500, receiver.produce().await.unwrap().unwrap_left());
                assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
            };

            join!(receive_things, send_things);
        });
    }

    #[test]
    fn test_spsc_capacity_1() {
        pollster::block_on(async {
            let (mut sender, mut receiver) = new_sssr(new_static::<i16, 1>());

            let send_things = async {
                assert!(sender.consume_item(300).await.is_ok());
                assert!(sender.consume_item(400).await.is_ok());
                assert!(sender.consume_item(500).await.is_ok());
                assert!(sender.consume_final(-17).await.is_ok());
            };

            let receive_things = async {
                assert_eq!(300, receiver.produce().await.unwrap().unwrap_left());
                assert_eq!(400, receiver.produce().await.unwrap().unwrap_left());
                assert_eq!(500, receiver.produce().await.unwrap().unwrap_left());
                assert_eq!(-17, receiver.produce().await.unwrap().unwrap_right());
            };

            join!(receive_things, send_things);
        });
    }
}