nuclei/sync/local/mpsc/
unbounded.rs

1//! The unbounded single-shard channel: a send never waits.
2
3use core::fmt;
4use std::rc::Rc;
5
6use super::chan::{Chan, Limit};
7use crate::sync::mpsc::error::SendError;
8
9pub(super) struct Unbounded;
10
11impl Limit for Unbounded {
12    fn release(&self, _n: usize) {}
13
14    fn is_idle(&self) -> bool {
15        true
16    }
17
18    fn close(&self) {}
19}
20
21/// Creates an unbounded channel and answers its two halves.
22///
23/// The halves belong to the tasks of one shard: they are `!Send`.
24///
25/// ```
26/// use nuclei::sync::local::mpsc;
27///
28/// let (tx, mut rx) = mpsc::unbounded_channel();
29/// tx.send(1).unwrap();
30/// tx.send(2).unwrap();
31/// assert_eq!(rx.try_recv(), Ok(1));
32/// assert_eq!(rx.try_recv(), Ok(2));
33/// ```
34pub fn unbounded_channel<T>() -> (UnboundedSender<T>, UnboundedReceiver<T>) {
35    let chan = Rc::new(Chan::new(Unbounded));
36    let tx = UnboundedSender {
37        chan: Rc::clone(&chan),
38    };
39    (tx, UnboundedReceiver { chan })
40}
41
42/// Sends values to the associated [`UnboundedReceiver`].
43///
44/// Created by [`unbounded_channel`]. Cloning it counts one more sender.
45pub struct UnboundedSender<T> {
46    chan: Rc<Chan<T, Unbounded>>,
47}
48
49impl<T> UnboundedSender<T> {
50    /// Sends a value; it never waits.
51    ///
52    /// # Errors
53    ///
54    /// A [`SendError`] holding the value when the receiver is closed or gone.
55    pub fn send(&self, value: T) -> Result<(), SendError<T>> {
56        if self.chan.is_closed() {
57            return Err(SendError(value));
58        }
59        self.chan.push(value);
60        Ok(())
61    }
62
63    /// Completes when the receiver has been closed or dropped. Cancel safe;
64    /// once the channel is closed every call completes at once.
65    pub async fn closed(&self) {
66        self.chan.closed().await;
67    }
68
69    /// Whether the receiver has been closed or dropped.
70    pub fn is_closed(&self) -> bool {
71        self.chan.is_closed()
72    }
73
74    /// Whether both senders send to the same channel.
75    pub fn same_channel(&self, other: &Self) -> bool {
76        Rc::ptr_eq(&self.chan, &other.chan)
77    }
78}
79
80impl<T> Clone for UnboundedSender<T> {
81    fn clone(&self) -> Self {
82        self.chan.add_sender();
83        Self {
84            chan: Rc::clone(&self.chan),
85        }
86    }
87}
88
89impl<T> Drop for UnboundedSender<T> {
90    fn drop(&mut self) {
91        self.chan.drop_sender();
92    }
93}
94
95impl<T> fmt::Debug for UnboundedSender<T> {
96    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
97        f.debug_struct("UnboundedSender")
98            .field("closed", &self.is_closed())
99            .finish()
100    }
101}
102
103/// Receives values from the associated [`UnboundedSender`]s.
104///
105/// Created by [`unbounded_channel`]. Dropping it closes the channel and drops
106/// the values still queued.
107pub struct UnboundedReceiver<T> {
108    chan: Rc<Chan<T, Unbounded>>,
109}
110
111impl<T> UnboundedReceiver<T> {
112    receiver_methods!();
113}
114
115impl<T> Drop for UnboundedReceiver<T> {
116    fn drop(&mut self) {
117        self.chan.drop_rx();
118    }
119}
120
121impl<T> fmt::Debug for UnboundedReceiver<T> {
122    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
123        f.debug_struct("UnboundedReceiver")
124            .field("len", &self.len())
125            .field("closed", &self.is_closed())
126            .finish()
127    }
128}
nuclei/sync/local/mpsc/unbounded.rs

nuclei/sync/local/mpsc/
unbounded.rs
