mecha10-core 0.8.1

//! Shared scaffolding for combinators that relay messages from one
//! [`Receiver`] into a freshly created one running on a background task.

use super::types::{Receiver, ReceiverInner};
use tokio::sync::mpsc;

pub(super) enum RelayAction<U> {
    Forward(U),
    Drop,
}

pub(super) enum RelaySender<T> {
    Unbounded(mpsc::UnboundedSender<T>),
    Bounded(mpsc::Sender<T>),
}

impl<T> RelaySender<T> {
    /// Send a value, awaiting backpressure on bounded channels.
    ///
    /// Returns `Err(())` if the corresponding receiver has been dropped.
    pub(super) async fn send(&self, value: T) -> std::result::Result<(), ()> {
        match self {
            RelaySender::Unbounded(tx) => tx.send(value).map_err(|_| ()),
            RelaySender::Bounded(tx) => tx.send(value).await.map_err(|_| ()),
        }
    }
}

/// Create a new channel whose boundedness (and capacity, if bounded) matches
/// `inner`, returning a sender/receiver-inner pair.
///
/// This is the shared "create a matching channel" scaffolding used by every
/// combinator that spawns a background task relaying messages from one
/// `Receiver` into a freshly created one (`filter`, `filter_map`, `debounce`,
/// `window`, ...).
pub(super) fn matching_channel<T, U>(inner: &ReceiverInner<T>) -> (RelaySender<U>, ReceiverInner<U>) {
    match inner {
        ReceiverInner::Unbounded(_) => {
            let (tx, rx) = mpsc::unbounded_channel();
            (RelaySender::Unbounded(tx), ReceiverInner::Unbounded(rx))
        }
        ReceiverInner::Bounded(bounded_rx) => {
            let capacity = bounded_rx.max_capacity();
            let (tx, rx) = mpsc::channel(capacity);
            (RelaySender::Bounded(tx), ReceiverInner::Bounded(rx))
        }
        ReceiverInner::DropOldest(drop_oldest_rx) => {
            let capacity = drop_oldest_rx.capacity();
            let (tx, rx) = mpsc::channel(capacity);
            (RelaySender::Bounded(tx), ReceiverInner::Bounded(rx))
        }
    }
}

impl<T> Receiver<T> {
    /// Create an output channel matching the boundedness (and, for bounded
    /// channels, the capacity) of `self`, then spawn a background task that
    /// pumps `self` through `transform` and forwards the result to the new
    /// channel.
    ///
    /// This is the shared scaffolding behind `filter()` and `filter_map()`:
    /// both are a "for each item, either forward it (possibly transformed)
    /// or drop it" loop, differing only in what `transform` decides for each
    /// item.
    pub(super) fn spawn_relay<U, F>(self, mut transform: F) -> Receiver<U>
    where
        F: FnMut(T) -> RelayAction<U> + Send + 'static,
        T: Send + 'static,
        U: Send + 'static,
    {
        let (tx, rx) = matching_channel::<T, U>(&self.inner);

        let mut receiver = self;
        tokio::spawn(async move {
            while let Some(msg) = receiver.recv().await {
                match transform(msg) {
                    RelayAction::Forward(item) => {
                        if tx.send(item).await.is_err() {
                            break;
                        }
                    }
                    RelayAction::Drop => {}
                }
            }
        });

        Receiver { inner: rx }
    }
}