tokio_broadcast/broadcast_map/
impl.rs

1use super::*;
2
3/// Constructs XXH3 state backed by the immutable default secret.
4impl Default for BroadcastMapHasher {
5    /// Initializes streaming state without allocating a secret buffer.
6    ///
7    /// # Returns
8    ///
9    /// - `BroadcastMapHasher` - A hasher with the original default seed and secret.
10    #[inline(always)]
11    fn default() -> Self {
12        Self(RawHasher::new(SecretBuffer::default()))
13    }
14}
15
16/// Delegates streaming operations to the dependency's allocation-free raw state.
17impl Hasher for BroadcastMapHasher {
18    /// Appends bytes to the streaming state.
19    ///
20    /// # Arguments
21    ///
22    /// - `&[u8]` - Bytes included in the hash.
23    ///
24    /// # Returns
25    ///
26    /// - `()` - Updates the current hash state.
27    #[inline(always)]
28    fn write(&mut self, input: &[u8]) {
29        self.0.write(input);
30    }
31
32    /// Computes the digest without consuming the streaming state.
33    ///
34    /// # Returns
35    ///
36    /// - `u64` - The XXH3 digest with the original default parameters.
37    #[inline(always)]
38    fn finish(&self) -> u64 {
39        self.0.finish()
40    }
41}
42
43/// Implements the `BroadcastMapTrait` for any type that also implements `Clone` and `Debug`.
44/// This blanket implementation allows any clonable and debuggable type to be used as a value in the broadcast map system.
45impl<T: Clone + Debug> BroadcastMapTrait for T {}
46
47/// Provides a default implementation for `BroadcastMap` instances.
48///
49/// The default broadcast map is initialized as an empty `DashMap`.
50impl<T: BroadcastMapTrait> Default for BroadcastMap<T> {
51    /// Creates a new, empty `BroadcastMap` instance.
52    ///
53    /// # Returns
54    ///
55    /// - `BroadcastMap<T>` - An empty broadcast map.
56    #[inline(always)]
57    fn default() -> Self {
58        Self(DashMap::with_hasher(BuildHasherDefault::default()))
59    }
60}
61
62/// Implements core functionalities for the `BroadcastMap` struct.
63impl<T: BroadcastMapTrait> BroadcastMap<T> {
64    /// Creates a new, empty `BroadcastMap` instance.
65    ///
66    /// This is a convenience constructor that simply calls `default()`.
67    ///
68    /// # Returns
69    ///
70    /// - `BroadcastMap<T>` - An empty broadcast map.
71    #[inline(always)]
72    pub fn new() -> Self {
73        Self::default()
74    }
75
76    /// Retrieves an immutable reference to the private channel storage.
77    ///
78    /// This private helper method provides direct access to the internal map.
79    ///
80    /// # Returns
81    ///
82    /// - `&BroadcastMapStorage<T>` - Reference to the internal map.
83    #[inline(always)]
84    fn get(&self) -> &BroadcastMapStorage<T> {
85        &self.0
86    }
87
88    /// Inserts a new broadcast channel into the map with a specified key and capacity.
89    ///
90    /// If a broadcast channel with the given key already exists, it will be replaced.
91    ///
92    /// # Arguments
93    ///
94    /// - `K` - Key convertible to `str`.
95    /// - `Capacity` - Maximum number of buffered messages.
96    ///
97    /// # Returns
98    ///
99    /// - `Option<Broadcast<T>>` - Previous broadcast channel if replaced.
100    #[inline(always)]
101    pub fn insert<K>(&self, key: K, capacity: Capacity) -> Option<Broadcast<T>>
102    where
103        K: AsRef<str>,
104    {
105        let broadcast: Broadcast<T> = Broadcast::new(capacity);
106        self.get().insert(key.as_ref().to_owned(), broadcast)
107    }
108
109    /// Retrieves the number of active receivers for the broadcast channel associated with the given key.
110    ///
111    /// # Arguments
112    ///
113    /// - `K` - Key convertible to `str`.
114    ///
115    /// # Returns
116    ///
117    /// - `Option<ReceiverCount>` - Number of receivers if channel exists.
118    #[inline(always)]
119    pub fn receiver_count<K>(&self, key: K) -> Option<ReceiverCount>
120    where
121        K: AsRef<str>,
122    {
123        self.get()
124            .get(key.as_ref())
125            .map(|receiver: Ref<'_, String, Broadcast<T>>| receiver.receiver_count())
126    }
127
128    /// Subscribes a new receiver to the broadcast channel associated with the given key.
129    ///
130    /// # Arguments
131    ///
132    /// - `K` - Key convertible to `str`.
133    ///
134    /// # Returns
135    ///
136    /// - `Option<BroadcastMapReceiver<T>>` - New receiver if channel exists.
137    #[inline(always)]
138    pub fn subscribe<K>(&self, key: K) -> Option<BroadcastMapReceiver<T>>
139    where
140        K: AsRef<str>,
141    {
142        self.get()
143            .get(key.as_ref())
144            .map(|receiver: Ref<'_, String, Broadcast<T>>| receiver.subscribe())
145    }
146
147    /// Subscribes a new receiver to the broadcast channel associated with the given key.
148    /// If the channel does not exist, it will be created with the specified capacity before subscribing.
149    ///
150    /// # Arguments
151    ///
152    /// - `K` - Key convertible to `str`.
153    /// - `Capacity` - Capacity for new channel if needed.
154    ///
155    /// # Returns
156    ///
157    /// - `BroadcastMapReceiver<T>` - New receiver for the channel.
158    #[inline(always)]
159    pub fn subscribe_or_insert<K>(&self, key: K, capacity: Capacity) -> BroadcastMapReceiver<T>
160    where
161        K: AsRef<str>,
162    {
163        let key_ref: &str = key.as_ref();
164        if let Some(sender) = self.get().get(key_ref) {
165            return sender.subscribe();
166        }
167        self.get()
168            .entry(key_ref.to_owned())
169            .or_insert_with(|| Broadcast::new(capacity))
170            .subscribe()
171    }
172
173    /// Attempts to send a message to the broadcast channel associated with the given key.
174    ///
175    /// # Arguments
176    ///
177    /// - `K` - Key convertible to `str`.
178    /// - `T` - Message to broadcast.
179    ///
180    /// # Returns
181    ///
182    /// - `Result<Option<ReceiverCount>, SendError<T>>` - Send result with receiver count or error.
183    #[inline(always)]
184    pub fn try_send<K>(&self, key: K, data: T) -> Result<Option<ReceiverCount>, SendError<T>>
185    where
186        K: AsRef<str>,
187    {
188        match self.get().get(key.as_ref()) {
189            Some(sender) => sender.send(data).map(Some),
190            None => Ok(None),
191        }
192    }
193
194    /// Sends a message to the broadcast channel associated with the given key.
195    ///
196    /// # Arguments
197    ///
198    /// - `K` - Key convertible to `str`.
199    /// - `T` - Message to broadcast.
200    ///
201    /// # Returns
202    ///
203    /// - `Option<ReceiverCount>` - The receiver count if the channel exists.
204    ///
205    /// # Panics
206    ///
207    /// Panics if the send operation fails (e.g., if the channel is closed).
208    #[inline(always)]
209    pub fn send<K>(&self, key: K, data: T) -> Option<ReceiverCount>
210    where
211        K: AsRef<str>,
212    {
213        self.try_send(key, data).unwrap()
214    }
215
216    /// Unsubscribes and removes the broadcast channel associated with the given key from the map.
217    ///
218    /// This operation effectively cancels all subscriptions to the channel by removing it from the map.
219    /// Any existing receivers will no longer receive new messages, and the channel will be dropped.
220    ///
221    /// # Arguments
222    ///
223    /// - `K` - Key convertible to `str`.
224    ///
225    /// # Returns
226    ///
227    /// - `Option<Broadcast<T>>` - The removed broadcast channel if it existed, or `None` if no channel was associated with the key.
228    #[inline(always)]
229    pub fn unsubscribe<K>(&self, key: K) -> Option<Broadcast<T>>
230    where
231        K: AsRef<str>,
232    {
233        self.get()
234            .remove(key.as_ref())
235            .map(|(_, broadcast): (String, Broadcast<T>)| broadcast)
236    }
237}
tokio_broadcast/broadcast_map/impl.rs

tokio_broadcast/broadcast_map/
impl.rs
