laminar_core/streaming/channel/
mod.rs

1//! Bounded MPSC channel backed by crossfire.
2
3use std::ops::Deref;
4
5use crossfire::mpsc;
6
7use super::config::ChannelConfig;
8use super::error::TryPushError;
9
10type Flavor<T> = mpsc::Array<T>;
11
12/// Sender half. Cloneable for multi-producer use.
13pub struct Producer<T: Send + 'static> {
14    tx: crossfire::MTx<Flavor<T>>,
15}
16
17/// Async receiver half.
18pub struct AsyncConsumer<T: Send + 'static> {
19    rx: crossfire::AsyncRx<Flavor<T>>,
20}
21
22/// Creates a bounded channel with blocking sender and async receiver.
23#[must_use]
24pub fn channel<T: Send + 'static>(buffer_size: usize) -> (Producer<T>, AsyncConsumer<T>) {
25    let cap = buffer_size.max(2);
26    let (tx, rx) = mpsc::bounded_blocking_async::<T>(cap);
27    (Producer { tx }, AsyncConsumer { rx })
28}
29
30/// Creates a bounded channel from a [`ChannelConfig`].
31#[must_use]
32pub(crate) fn channel_with_config<T: Send + 'static>(
33    config: &ChannelConfig,
34) -> (Producer<T>, AsyncConsumer<T>) {
35    channel(config.buffer_size)
36}
37
38// -- Producer -----------------------------------------------------------------
39
40impl<T: Send + 'static> Producer<T> {
41    /// Non-blocking send.
42    ///
43    /// # Errors
44    ///
45    /// Returns the item if the channel is full or the receiver was dropped.
46    pub fn push(&self, item: T) -> Result<(), T> {
47        self.tx
48            .try_send(item)
49            .map_err(crossfire::TrySendError::into_inner)
50    }
51
52    /// Non-blocking send with typed error.
53    ///
54    /// # Errors
55    ///
56    /// Returns `TryPushError` containing the item if the channel is full.
57    #[inline]
58    pub fn try_push(&self, item: T) -> Result<(), TryPushError<T>> {
59        self.tx.try_send(item).map_err(|e| match e {
60            crossfire::TrySendError::Full(v) => TryPushError::full(v),
61            crossfire::TrySendError::Disconnected(v) => TryPushError::disconnected(v),
62        })
63    }
64
65    /// Returns `true` if the receiver has been dropped.
66    #[must_use]
67    pub fn is_closed(&self) -> bool {
68        self.tx.is_disconnected()
69    }
70
71    /// Number of items currently buffered.
72    #[must_use]
73    pub fn len(&self) -> usize {
74        self.tx.deref().len()
75    }
76
77    /// Buffer capacity.
78    #[must_use]
79    pub fn capacity(&self) -> usize {
80        self.tx.deref().capacity().unwrap_or(0)
81    }
82
83    /// Whether the buffer is empty.
84    #[must_use]
85    pub fn is_empty(&self) -> bool {
86        self.tx.deref().is_empty()
87    }
88}
89
90impl<T: Send + 'static> Clone for Producer<T> {
91    fn clone(&self) -> Self {
92        Self {
93            tx: self.tx.clone(),
94        }
95    }
96}
97
98impl<T: Send + 'static> std::fmt::Debug for Producer<T> {
99    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100        f.debug_struct("Producer")
101            .field("len", &self.len())
102            .field("capacity", &self.capacity())
103            .finish()
104    }
105}
106
107// -- AsyncConsumer ------------------------------------------------------------
108
109impl<T: Send + 'static> AsyncConsumer<T> {
110    /// Async receive. Suspends until a message arrives or the sender disconnects.
111    ///
112    /// # Errors
113    ///
114    /// Returns `crossfire::RecvError` if the sender was dropped.
115    pub async fn recv(&mut self) -> Result<T, crossfire::RecvError> {
116        self.rx.recv().await
117    }
118
119    /// Non-blocking receive. Returns `Err` immediately if the channel is
120    /// empty or the senders have all disconnected.
121    ///
122    /// # Errors
123    ///
124    /// Returns `crossfire::TryRecvError::Empty` when no items are buffered,
125    /// `crossfire::TryRecvError::Disconnected` after all senders are dropped
126    /// and the buffer is drained.
127    pub fn try_recv(&self) -> Result<T, crossfire::TryRecvError> {
128        self.rx.try_recv()
129    }
130
131    /// Returns `true` if the sender has been dropped.
132    #[must_use]
133    pub fn is_disconnected(&self) -> bool {
134        self.rx.is_disconnected()
135    }
136}
137
138impl<T: Send + 'static> std::fmt::Debug for AsyncConsumer<T> {
139    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
140        f.debug_struct("AsyncConsumer")
141            .field("disconnected", &self.is_disconnected())
142            .finish()
143    }
144}
145
146#[cfg(test)]
147mod tests;
laminar_core/streaming/channel/mod.rs

laminar_core/streaming/channel/
mod.rs
