ufotofu 0.12.1

//! A **S**ingle-**S**ender **S**ingle-**R**eceiver FIFO channel.
//!
//! The entrypoint to this module is the [`new_sssr`] function. You call it with a reference to a [`State`], 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 buffer 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
//!
//! An example with a stack-allocated [`State`]:
//!
//! ```
//! use futures::join;
//! use ufotofu::prelude::*;
//! use ufotofu::channels::advanced::sssr::*;
//!
//! // Allocate a new opaque state on the stack.
//! let state = State::new(ufotofu::queues::new_static::<i16, 2>());
//!
//! // Create a channel whose endpoints reference the shared state through vanilla reference.
//! // The state must outlive the endpoints, otherwise you would get a compiler error.
//! let (mut sender, mut receiver) = new_sssr(&state);
//!
//! 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::{
    cell::Cell,
    convert::Infallible,
    fmt,
    ops::{Deref, DerefMut},
};

use either::Either::{self, *};

use fairly_unsafe_cell::*;
use frugal_async::{Mutex, TakeCell};

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

/// The state shared between the senders and receivers of an sssr in-memory channel.
///
/// This type is fully opaque, we expose it merely to give control over where it is allocated.
///
/// `Q` is the type of the internal item queue, `F` is the type of final items for the channel.
pub struct State<Q, F> {
    // We need a Mutex here because `expose_slots` and `expose_items` can be called concurrently on the two endpoints.
    queue: Mutex<Q>,
    // Safety: We never return refs to this from any method, and we never hold a borrow across `.await` points.
    // Hence, no concurrent refs can exist.
    buffered_final_value: FairlyUnsafeCell<Option<F>>,
    // We track the number of items in the queue here, so that we can access it without waiting for the Mutex of the queue itself.
    // A bit awkward, but this enables sync access to the current count.
    len: Cell<usize>,
    // Empty while the sender cannot make progress.
    notify_the_sender: TakeCell<()>,
    // Empty while the receiver cannot make progress.
    notify_the_receiver: TakeCell<()>,
    // True iff neither endpoint has been dropped yet.
    did_any_endpoint_drop_yet: Cell<bool>,
}

impl<Q, F> fmt::Debug for State<Q, F>
where
    Q: fmt::Debug,
    F: fmt::Debug,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("State")
            .field("queue_item_count", &self.len)
            .field("queue", &self.queue)
            .field("buffered_final_value", &self.buffered_final_value)
            .finish()
    }
}

impl<Q: Queue, F> State<Q, F> {
    /// Creates a new [`State`], using the given queue as backing storage for the channel.
    pub fn new(queue: Q) -> Self {
        State {
            len: Cell::new(queue.len()),
            queue: Mutex::new(queue),
            buffered_final_value: FairlyUnsafeCell::new(None),
            notify_the_sender: TakeCell::new_with(()),
            notify_the_receiver: TakeCell::new(),
            did_any_endpoint_drop_yet: Cell::new(false),
        }
    }

    /// Returns the number of items in the queue that buffers items for this channel.
    fn len(&self) -> usize {
        self.len.get()
    }

    /// Returns whether the queue that buffers items for this channel is currently empty.
    fn is_empty(&self) -> bool {
        self.len.get() == 0
    }

    /// Performs the actual sequence-related logic of closing the channel.
    fn close(&self, fin: F) {
        // Store the final value for later access by the Senders.
        let mut last = unsafe { self.buffered_final_value.borrow_mut() };
        *last = Some(fin);

        self.notify_the_receiver.set(());
    }
}

/// Creates a new single-sender single-receiver channel in the form of a [`Sender`] and a [`Receiver`] endpoint, which communicate via the given [`State`].
///
/// See [`ufotofu::channels::new_sssr`](crate::channels::new_sssr) for an API that hides the explicit [`State`] management (by transparently allocating the state on the heap and freeing it automatically after both endpoints are dropped).
///
/// #### Example
///
/// An example with a stack-allocated [`State`]:
///
/// ```
/// use futures::join;
/// use ufotofu::prelude::*;
/// use ufotofu::channels::advanced::sssr::*;
///
/// // Allocate a new opaque state on the stack.
/// let state = State::new(ufotofu::queues::new_static::<i16, 2>());
///
/// // Create a channel whose endpoints reference the shared state through vanilla reference.
/// // The state must outlive the endpoints, otherwise you would get a compiler error.
/// let (mut sender, mut receiver) = new_sssr(&state);
///
/// 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<R, Q, F>(state_ref: R) -> (Sender<R, Q, F>, Receiver<R, Q, F>)
where
    R: Deref<Target = State<Q, F>> + Clone,
{
    (
        Sender {
            state: state_ref.clone(),
        },
        Receiver { state: state_ref },
    )
}

/// 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.
#[derive(Debug)]
pub struct Sender<R, Q, F>
where
    R: Deref<Target = State<Q, F>>,
{
    state: R,
}

impl<R, Q, F> Drop for Sender<R, Q, F>
where
    R: Deref<Target = State<Q, F>>,
{
    fn drop(&mut self) {
        self.state.did_any_endpoint_drop_yet.set(true)
    }
}

impl<R, Q, F> Sender<R, Q, F>
where
    R: Deref<Target = State<Q, F>>,
    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::advanced::sssr::*;
    ///
    /// let state = State::new(ufotofu::queues::new_static::<i16, 2>());
    /// let (mut sender, mut receiver) = new_sssr(&state);
    ///
    /// 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.state.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::advanced::sssr::*;
    ///
    /// let state = State::new(ufotofu::queues::new_static::<i16, 2>());
    /// let (mut sender, mut receiver) = new_sssr(&state);
    ///
    /// 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.state.is_empty()
    }

    /// Returns whether the corresponding [`Receiver`] has been dropped already.
    ///
    /// ```
    /// use ufotofu::channels::advanced::sssr::*;
    ///
    /// let state: State<_, ()> = State::new(ufotofu::queues::new_static::<i16, 2>());
    /// let (sender, receiver) = new_sssr(&state);
    ///
    /// 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.state.did_any_endpoint_drop_yet.get()
    }
}

impl<R: Deref<Target = State<Q, F>>, Q: Queue, F> Consumer for Sender<R, 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> {
        match val {
            Left(mut item) => {
                loop {
                    // Try to buffer the item.
                    let did_it_work = {
                        // Inside a block to drop the Mutex access before awaiting on the notifier.
                        self.state.queue.write().await.deref_mut().enqueue(item)
                    };

                    match did_it_work {
                        // Enqueueing failed.
                        Some(item_) => {
                            // Wait for queue space.
                            let () = self.state.notify_the_sender.take().await;
                            // Go into the next iteration of the loop, where enqueeuing is guaranteed to succeed.
                            item = item_;
                        }
                        // Enqueueing succeeded.
                        None => {
                            self.state.len.set(self.state.len.get() + 1);
                            self.state.notify_the_receiver.set(());
                            return Ok(());
                        }
                    }
                }
            }

            Right(fin) => {
                self.state.close(fin);
                Ok(())
            }
        }
    }

    async fn flush(&mut self) -> Result<(), Self::Error> {
        Ok(()) // Nothing to do.
    }
}

impl<R: Deref<Target = State<Q, F>>, Q: Queue, F> BulkConsumer for Sender<R, 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),
    {
        let mut f = Some(f);

        loop {
            let ret = self
                .state
                .queue
                .write()
                .await
                .deref_mut()
                .expose_slots(async |queue_slots| {
                    if queue_slots.is_empty() {
                        (0, None)
                    } else {
                        let (amount, ret) = (f.take().expect(
                            "Running this branch only once, we return after having called f",
                        ))(queue_slots)
                        .await;
                        self.state.len.set(self.state.len.get() + amount);
                        self.state.notify_the_receiver.set(());
                        (amount, Some(ret))
                    }
                })
                .await;

            match ret {
                None => {
                    let () = self.state.notify_the_sender.take().await;
                    // And now go into the next iteration of the loop, where there will be slots available.
                }
                Some(ret) => return Ok(ret),
            }
        }
    }
}

/// 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.
#[derive(Debug)]
pub struct Receiver<R, Q, F>
where
    R: Deref<Target = State<Q, F>>,
{
    state: R,
}

impl<R, Q, F> Drop for Receiver<R, Q, F>
where
    R: Deref<Target = State<Q, F>>,
{
    fn drop(&mut self) {
        self.state.did_any_endpoint_drop_yet.set(true)
    }
}

impl<R: Deref<Target = State<Q, F>>, Q: Queue, F> Receiver<R, 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::advanced::sssr::*;
    ///
    /// let state = State::new(ufotofu::queues::new_static::<i16, 2>());
    /// let (mut sender, mut receiver) = new_sssr(&state);
    ///
    /// 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.state.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::advanced::sssr::*;
    ///
    /// let state = State::new(ufotofu::queues::new_static::<i16, 2>());
    /// let (mut sender, mut receiver) = new_sssr(&state);
    ///
    /// 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.state.is_empty()
    }

    /// Returns whether the corresponding [`Sender`] has been dropped already.
    ///
    /// ```
    /// use ufotofu::channels::advanced::sssr::*;
    ///
    /// let state: State<_, ()> = State::new(ufotofu::queues::new_static::<i16, 2>());
    /// let (sender, receiver) = new_sssr(&state);
    ///
    /// 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.state.did_any_endpoint_drop_yet.get()
    }
}

impl<R: Deref<Target = State<Q, F>>, Q: Queue, F> Producer for Receiver<R, 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> {
        loop {
            // Try to obtain the next item.
            match self.state.queue.write().await.deref_mut().dequeue() {
                // At least one item was in the buffer, return it.
                Some(item) => {
                    self.state.len.set(self.state.len.get() - 1);
                    self.state.notify_the_sender.set(());
                    return Ok(Left(item));
                }
                None => {
                    // Buffer is empty.
                    // But perhaps the final item has been consumed already, or an error was requested?
                    match unsafe { self.state.buffered_final_value.borrow_mut().take() } {
                        Some(fin) => {
                            return Ok(Right(fin));
                        }
                        None => {
                            // No last item yet, so we wait until something changes.
                            // But we do the waiting outside this `match` block, so that the mutex is released first.
                        }
                    }
                }
            }

            // No last item yet, so we wait until something changes.
            let () = self.state.notify_the_receiver.take().await;
            // Go into the next iteration of the loop, where progress will be made.
        }
    }

    async fn slurp(&mut self) -> Result<(), Self::Error> {
        Ok(()) // Nothing to do.
    }
}

impl<R: Deref<Target = State<Q, F>>, Q: Queue, F> BulkProducer for Receiver<R, 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),
    {
        let mut f = Some(f);

        loop {
            let ret = self
                .state
                .queue
                .write()
                .await
                .expose_items(async |queue_items| {
                    if queue_items.is_empty() {
                        match unsafe { self.state.buffered_final_value.borrow_mut().take() } {
                            Some(fin) => (0, Some(Right(fin))),
                            None => (0, None),
                        }
                    } else {
                        let (amount, ret) = (f.take().expect(
                            "Running this branch only once, we return after having called f",
                        ))(queue_items)
                        .await;
                        self.state.len.set(self.state.len.get() - amount);
                        self.state.notify_the_sender.set(());
                        (amount, Some(Left(ret)))
                    }
                })
                .await;

            match ret {
                None => {
                    let () = self.state.notify_the_receiver.take().await;
                    // And now go into the next iteration of the loop, where there will be slots available.
                }
                Some(Left(ret)) => return Ok(Left(ret)),
                Some(Right(fin)) => {
                    return Ok(Right((
                        f.take()
                            .expect("Branch guarded by early return of Ok(Left(_))"),
                        fin,
                    )))
                }
            }
        }
    }
}

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

    use futures::join;

    use crate::queues::new_static;

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

        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 state = State::new(new_static::<i16, 2>());
            let (mut sender, mut receiver) = new_sssr(&state);

            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 state = State::new(new_static::<i16, 3>());
            let (mut sender, mut receiver) = new_sssr(&state);

            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 state = State::new(new_static::<i16, 2>());
            let (mut sender, mut receiver) = new_sssr(&state);

            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 state = State::new(new_static::<i16, 1>());
            let (mut sender, mut receiver) = new_sssr(&state);

            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);
        });
    }
}