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    pub fn try_push(&self, item: T) -> Result<(), TryPushError<T>> {
58        self.tx.try_send(item).map_err(|e| match e {
59            crossfire::TrySendError::Full(v) => TryPushError::full(v),
60            crossfire::TrySendError::Disconnected(v) => TryPushError::disconnected(v),
61        })
62    }
63
64    /// Returns `true` if the receiver has been dropped.
65    #[must_use]
66    pub fn is_closed(&self) -> bool {
67        self.tx.is_disconnected()
68    }
69
70    /// Number of items currently buffered.
71    #[must_use]
72    pub fn len(&self) -> usize {
73        self.tx.deref().len()
74    }
75
76    /// Buffer capacity.
77    #[must_use]
78    pub fn capacity(&self) -> usize {
79        self.tx.deref().capacity().unwrap_or(0)
80    }
81
82    /// Whether the buffer is empty.
83    #[must_use]
84    pub fn is_empty(&self) -> bool {
85        self.tx.deref().is_empty()
86    }
87}
88
89impl<T: Send + 'static> Clone for Producer<T> {
90    fn clone(&self) -> Self {
91        Self {
92            tx: self.tx.clone(),
93        }
94    }
95}
96
97impl<T: Send + 'static> std::fmt::Debug for Producer<T> {
98    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
99        f.debug_struct("Producer")
100            .field("len", &self.len())
101            .field("capacity", &self.capacity())
102            .finish()
103    }
104}
105
106// -- AsyncConsumer ------------------------------------------------------------
107
108impl<T: Send + 'static> AsyncConsumer<T> {
109    /// Async receive. Suspends until a message arrives or the sender disconnects.
110    ///
111    /// # Errors
112    ///
113    /// Returns `crossfire::RecvError` if the sender was dropped.
114    pub async fn recv(&mut self) -> Result<T, crossfire::RecvError> {
115        self.rx.recv().await
116    }
117
118    /// Non-blocking receive. Returns `Err` immediately if the channel is
119    /// empty or the senders have all disconnected.
120    ///
121    /// # Errors
122    ///
123    /// Returns `crossfire::TryRecvError::Empty` when no items are buffered,
124    /// `crossfire::TryRecvError::Disconnected` after all senders are dropped
125    /// and the buffer is drained.
126    pub fn try_recv(&self) -> Result<T, crossfire::TryRecvError> {
127        self.rx.try_recv()
128    }
129
130    /// Returns `true` if the sender has been dropped.
131    #[must_use]
132    pub fn is_disconnected(&self) -> bool {
133        self.rx.is_disconnected()
134    }
135}
136
137impl<T: Send + 'static> std::fmt::Debug for AsyncConsumer<T> {
138    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
139        f.debug_struct("AsyncConsumer")
140            .field("disconnected", &self.is_disconnected())
141            .finish()
142    }
143}
144
145#[cfg(test)]
146mod tests;
laminar_core/streaming/channel/mod.rs

laminar_core/streaming/channel/
mod.rs
