medea_reactive/collections/
hash_map.rs

1//! Reactive hash map backed by [`HashMap`][1].
2//!
3//! [1]: std::collections::HashMap
4
5use std::{
6    collections::{
7        HashMap as StdHashMap,
8        hash_map::{Iter, Values},
9    },
10    hash::Hash,
11    marker::PhantomData,
12};
13
14use futures::stream::{self, LocalBoxStream, StreamExt as _};
15
16use crate::subscribers_store::{
17    SubscribersStore, common, progressable,
18    progressable::{AllProcessed, Processed},
19};
20
21/// Reactive hash map based on [`HashMap`][1] with additional functionality of
22/// tracking progress made by its subscribers.
23///
24/// Its [`HashMap::on_insert()`] and [`HashMap::on_remove()`] subscriptions
25/// return values wrapped in [`progressable::Guarded`], and implementation
26/// tracks all [`progressable::Guard`]s.
27///
28/// [1]: std::collections::HashMap
29pub type ProgressableHashMap<K, V> = HashMap<
30    K,
31    V,
32    progressable::SubStore<(K, V)>,
33    progressable::Guarded<(K, V)>,
34>;
35
36/// Reactive hash map based on [`HashMap`].
37pub type ObservableHashMap<K, V> =
38    HashMap<K, V, common::SubStore<(K, V)>, (K, V)>;
39
40/// Reactive hash map based on [`HashMap`].
41///
42/// # Usage
43///
44/// ```rust
45/// # use std::collections::HashMap;
46/// # use futures::{executor, StreamExt as _};
47/// use medea_reactive::collections::ObservableHashMap;
48///
49/// # executor::block_on(async {
50/// let mut map = ObservableHashMap::new();
51///
52/// // You can subscribe on insert action:
53/// let mut inserts = map.on_insert();
54/// map.insert("foo", "bar");
55/// let (key, val) = inserts.next().await.unwrap();
56/// assert_eq!(key, "foo");
57/// assert_eq!(val, "bar");
58///
59/// // Also you can subscribe on remove action:
60/// let mut removals = map.on_remove();
61/// map.remove(&"foo");
62/// let (key, val) = removals.next().await.unwrap();
63/// assert_eq!(key, "foo");
64/// assert_eq!(val, "bar");
65///
66/// // Remove subscription will also receive all items of the HashMap when it
67/// // will be dropped:
68/// map.insert("foo-1", "bar-1");
69/// map.insert("foo-2", "bar-2");
70/// drop(map);
71/// let removed_items: HashMap<_, _> = removals.take(2).collect().await;
72/// assert_eq!(removed_items["foo-1"], "bar-1");
73/// assert_eq!(removed_items["foo-2"], "bar-2");
74/// # });
75/// ```
76///
77/// # Waiting for subscribers to complete
78///
79/// ```rust
80/// # use futures::{executor, StreamExt as _, Stream};
81/// use medea_reactive::collections::ProgressableHashMap;
82///
83/// # executor::block_on(async {
84/// let mut hash_map = ProgressableHashMap::new();
85///
86/// let mut on_insert = hash_map.on_insert();
87/// hash_map.insert(1, 1);
88///
89/// // hash_map.when_insert_processed().await; <- wouldn't be resolved
90/// let value = on_insert.next().await.unwrap();
91/// // hash_map.when_insert_processed().await; <- wouldn't be resolved
92/// drop(value);
93///
94/// hash_map.when_insert_processed().await; // will be resolved
95///
96/// # });
97/// ```
98#[derive(Debug, Clone)]
99pub struct HashMap<K, V, S: SubscribersStore<(K, V), O>, O> {
100    /// Data stored by this [`HashMap`].
101    store: StdHashMap<K, V>,
102
103    /// Subscribers of the [`HashMap::on_insert()`] method.
104    on_insert_subs: S,
105
106    /// Subscribers of the [`HashMap::on_remove()`] method.
107    on_remove_subs: S,
108
109    /// Phantom type of [`HashMap::on_insert()`] and [`HashMap::on_remove()`]
110    /// output.
111    _output: PhantomData<O>,
112}
113
114impl<K, V> ProgressableHashMap<K, V>
115where
116    K: Hash + Eq + Clone + 'static,
117    V: Clone + 'static,
118{
119    /// Returns [`Future`] resolving when all insertion updates will be
120    /// processed by [`HashMap::on_insert()`] subscribers.
121    pub fn when_insert_processed(&self) -> Processed<'static> {
122        self.on_insert_subs.when_all_processed()
123    }
124
125    /// Returns [`Future`] resolving when all remove updates will be processed
126    /// by [`HashMap::on_remove()`] subscribers.
127    pub fn when_remove_processed(&self) -> Processed<'static> {
128        self.on_remove_subs.when_all_processed()
129    }
130
131    /// Returns [`Future`] resolving when all insert and remove updates will be
132    /// processed by subscribers.
133    pub fn when_all_processed(&self) -> AllProcessed<'static> {
134        crate::when_all_processed(vec![
135            self.when_remove_processed().into(),
136            self.when_insert_processed().into(),
137        ])
138    }
139}
140
141impl<K, V, S: SubscribersStore<(K, V), O>, O> HashMap<K, V, S, O> {
142    /// Creates new empty [`HashMap`].
143    #[must_use]
144    pub fn new() -> Self {
145        Self::default()
146    }
147
148    /// [`Iterator`] visiting all key-value pairs in an arbitrary order.
149    pub fn iter(&self) -> impl Iterator<Item = (&K, &V)> {
150        self.into_iter()
151    }
152
153    /// [`Iterator`] visiting all values in an arbitrary order.
154    #[must_use]
155    pub fn values(&self) -> Values<'_, K, V> {
156        self.store.values()
157    }
158
159    /// Returns [`Stream`] yielding inserted key-value pairs to this
160    /// [`HashMap`].
161    ///
162    /// [`Stream`]: futures::Stream
163    #[must_use]
164    pub fn on_insert(&self) -> LocalBoxStream<'static, O> {
165        self.on_insert_subs.subscribe()
166    }
167
168    /// Returns [`Stream`] yielding removed key-value pairs from this
169    /// [`HashMap`].
170    ///
171    /// Note, that this [`Stream`] will yield all key-value pairs of this
172    /// [`HashMap`] on [`Drop`].
173    ///
174    /// [`Stream`]: futures::Stream
175    #[must_use]
176    pub fn on_remove(&self) -> LocalBoxStream<'static, O> {
177        self.on_remove_subs.subscribe()
178    }
179}
180
181impl<K, V, S, O> HashMap<K, V, S, O>
182where
183    K: Clone,
184    V: Clone,
185    S: SubscribersStore<(K, V), O>,
186    O: 'static,
187{
188    /// Returns [`Stream`] containing values from this [`HashMap`].
189    ///
190    /// Returned [`Stream`] contains only current values. It won't update on new
191    /// inserts, but you can merge returned [`Stream`] with a
192    /// [`HashMap::on_insert()`] [`Stream`] if you want to process current
193    /// values and values that will be inserted.
194    ///
195    /// [`Stream`]: futures::Stream
196    #[expect(clippy::needless_collect, reason = "false positive: lifetimes")]
197    pub fn replay_on_insert(&self) -> LocalBoxStream<'static, O> {
198        Box::pin(stream::iter(
199            self.store
200                .iter()
201                .map(|(k, v)| self.on_insert_subs.wrap((k.clone(), v.clone())))
202                .collect::<Vec<_>>(),
203        ))
204    }
205
206    /// Chains [`HashMap::replay_on_insert()`] with a [`HashMap::on_insert()`].
207    pub fn on_insert_with_replay(&self) -> LocalBoxStream<'static, O> {
208        Box::pin(self.replay_on_insert().chain(self.on_insert()))
209    }
210}
211
212impl<K, V, S, O> HashMap<K, V, S, O>
213where
214    K: Hash + Eq,
215    S: SubscribersStore<(K, V), O>,
216{
217    /// Returns a reference to the value corresponding to the `key`.
218    #[must_use]
219    pub fn get(&self, key: &K) -> Option<&V> {
220        self.store.get(key)
221    }
222
223    /// Returns a mutable reference to the value corresponding to the `key`.
224    ///
225    /// Note, that mutating of the returned value wouldn't work same as
226    /// [`Observable`]s and doesn't spawns [`HashMap::on_insert()`] or
227    /// [`HashMap::on_remove()`] events. If you need subscriptions on value
228    /// changes then just wrap the value into an [`Observable`] and subscribe to
229    /// it.
230    ///
231    /// [`Observable`]: crate::Observable
232    #[must_use]
233    pub fn get_mut(&mut self, key: &K) -> Option<&mut V> {
234        self.store.get_mut(key)
235    }
236}
237
238impl<K, V, S, O> HashMap<K, V, S, O>
239where
240    K: Hash + Eq + Clone,
241    V: Clone,
242    S: SubscribersStore<(K, V), O>,
243{
244    /// Removes all entries which are not present in the provided [`HashMap`].
245    ///
246    /// [`HashMap`]: std::collections::HashMap
247    pub fn remove_not_present<A>(&mut self, other: &StdHashMap<K, A>) {
248        self.iter()
249            .filter_map(|(id, _)| {
250                if other.contains_key(id) { None } else { Some(id.clone()) }
251            })
252            .collect::<Vec<_>>()
253            .into_iter()
254            .for_each(|id| drop(self.remove(&id)));
255    }
256
257    /// Inserts a key-value pair to this [`HashMap`].
258    ///
259    /// Emits [`HashMap::on_insert()`] event and may emit
260    /// [`HashMap::on_remove()`] event if insert replaces a value contained in
261    /// this [`HashMap`].
262    pub fn insert(&mut self, key: K, value: V) -> Option<V> {
263        let removed_value = self.store.insert(key.clone(), value.clone());
264        if let Some(val) = &removed_value {
265            self.on_remove_subs.send_update((key.clone(), val.clone()));
266        }
267
268        self.on_insert_subs.send_update((key, value));
269
270        removed_value
271    }
272
273    /// Removes the `key` from this [`HashMap`], returning the value behind it,
274    /// if any.
275    ///
276    /// Emits [`HashMap::on_remove()`] event if value with provided key is
277    /// removed from this [`HashMap`].
278    pub fn remove(&mut self, key: &K) -> Option<V> {
279        let removed_item = self.store.remove(key);
280        if let Some(item) = &removed_item {
281            self.on_remove_subs.send_update((key.clone(), item.clone()));
282        }
283
284        removed_item
285    }
286}
287
288// Implemented manually to omit redundant `: Default` trait bounds, imposed by
289// `#[derive(Default)]`.
290impl<K, V, S: SubscribersStore<(K, V), O>, O> Default for HashMap<K, V, S, O> {
291    fn default() -> Self {
292        Self {
293            store: StdHashMap::new(),
294            on_insert_subs: S::default(),
295            on_remove_subs: S::default(),
296            _output: PhantomData,
297        }
298    }
299}
300
301impl<K, V, S: SubscribersStore<(K, V), O>, O> From<StdHashMap<K, V>>
302    for HashMap<K, V, S, O>
303{
304    fn from(from: StdHashMap<K, V>) -> Self {
305        Self {
306            store: from,
307            on_remove_subs: S::default(),
308            on_insert_subs: S::default(),
309            _output: PhantomData,
310        }
311    }
312}
313
314impl<'a, K, V, S: SubscribersStore<(K, V), O>, O> IntoIterator
315    for &'a HashMap<K, V, S, O>
316{
317    type IntoIter = Iter<'a, K, V>;
318    type Item = (&'a K, &'a V);
319
320    fn into_iter(self) -> Self::IntoIter {
321        self.store.iter()
322    }
323}
324
325impl<K, V, S: SubscribersStore<(K, V), O>, O> Drop for HashMap<K, V, S, O> {
326    /// Sends all key-values of a dropped [`HashMap`] to the
327    /// [`HashMap::on_remove`] subs.
328    fn drop(&mut self) {
329        #[expect(clippy::iter_over_hash_type, reason = "order doesn't matte")]
330        for (k, v) in self.store.drain() {
331            self.on_remove_subs.send_update((k, v));
332        }
333    }
334}
335
336impl<K, V, S: SubscribersStore<(K, V), O>, O> FromIterator<(K, V)>
337    for HashMap<K, V, S, O>
338where
339    K: Hash + Eq,
340{
341    fn from_iter<T: IntoIterator<Item = (K, V)>>(iter: T) -> Self {
342        Self {
343            store: StdHashMap::from_iter(iter),
344            on_remove_subs: S::default(),
345            on_insert_subs: S::default(),
346            _output: PhantomData,
347        }
348    }
349}
350
351#[cfg(test)]
352mod tests {
353    use futures::{FutureExt as _, StreamExt as _, poll, task::Poll};
354
355    use crate::collections::ProgressableHashMap;
356
357    #[tokio::test]
358    async fn replace_triggers_on_remove() {
359        let mut map = ProgressableHashMap::new();
360        let _ = map.insert(0u32, 0u32);
361
362        let mut on_insert = map.on_insert();
363        let mut on_remove = map.on_remove();
364
365        assert_eq!(map.insert(0, 1).unwrap(), 0);
366
367        assert_eq!(*on_insert.next().await.unwrap(), (0, 1));
368        assert_eq!(*on_remove.next().await.unwrap(), (0, 0));
369    }
370
371    #[tokio::test]
372    async fn replay_on_insert() {
373        let mut map = ProgressableHashMap::new();
374
375        let _ = map.insert(0, 0);
376        let _ = map.insert(1, 2);
377        let _ = map.insert(1, 2);
378        let _ = map.insert(2, 3);
379
380        let inserts: Vec<_> =
381            map.replay_on_insert().map(|val| val.into_inner()).collect().await;
382
383        assert_eq!(inserts.len(), 3);
384        assert!(inserts.contains(&(0, 0)));
385        assert!(inserts.contains(&(1, 2)));
386        assert!(inserts.contains(&(2, 3)));
387    }
388
389    #[tokio::test]
390    async fn when_remove_processed() {
391        let mut map = ProgressableHashMap::new();
392        let _ = map.insert(0, 0);
393
394        let mut on_remove = map.on_remove();
395
396        assert_eq!(poll!(map.when_remove_processed()), Poll::Ready(()));
397        assert_eq!(map.remove(&0), Some(0));
398        assert_eq!(poll!(map.when_remove_processed()), Poll::Pending);
399
400        let (val, guard) = on_remove.next().await.unwrap().into_parts();
401
402        assert_eq!(val, (0, 0));
403        assert_eq!(poll!(map.when_remove_processed()), Poll::Pending);
404        drop(guard);
405        assert_eq!(poll!(map.when_remove_processed()), Poll::Ready(()));
406    }
407
408    #[tokio::test]
409    async fn multiple_when_remove_processed_subs() {
410        let mut map = ProgressableHashMap::new();
411        let _ = map.insert(0, 0);
412
413        let mut on_remove1 = map.on_remove();
414        let mut on_remove2 = map.on_remove();
415
416        assert_eq!(poll!(map.when_remove_processed()), Poll::Ready(()));
417        let _ = map.remove(&0).unwrap();
418        assert_eq!(poll!(map.when_remove_processed()), Poll::Pending);
419
420        assert_eq!(on_remove1.next().await.unwrap().into_inner(), (0, 0));
421        assert_eq!(poll!(map.when_remove_processed()), Poll::Pending);
422        assert_eq!(on_remove2.next().await.unwrap().into_inner(), (0, 0));
423
424        assert_eq!(poll!(map.when_remove_processed()), Poll::Ready(()));
425    }
426
427    #[tokio::test]
428    async fn when_insert_processed() {
429        let mut map = ProgressableHashMap::new();
430        let _ = map.insert(0, 0);
431
432        let mut on_insert = map.on_insert();
433
434        assert_eq!(poll!(map.when_insert_processed()), Poll::Ready(()));
435        let _ = map.insert(2, 3);
436        assert_eq!(poll!(map.when_insert_processed()), Poll::Pending);
437
438        let (val, guard) = on_insert.next().await.unwrap().into_parts();
439
440        assert_eq!(val, (2, 3));
441        assert_eq!(poll!(map.when_insert_processed()), Poll::Pending);
442        drop(guard);
443        assert_eq!(poll!(map.when_insert_processed()), Poll::Ready(()));
444    }
445
446    #[tokio::test]
447    async fn multiple_when_insert_processed_subs() {
448        let mut map = ProgressableHashMap::new();
449        let _ = map.insert(0, 0);
450
451        let mut on_insert1 = map.on_insert();
452        let mut on_insert2 = map.on_insert();
453
454        assert_eq!(poll!(map.when_insert_processed()), Poll::Ready(()));
455        let _ = map.insert(0, 0).unwrap();
456        assert_eq!(poll!(map.when_insert_processed()), Poll::Pending);
457
458        assert_eq!(on_insert1.next().await.unwrap().into_inner(), (0, 0));
459        assert_eq!(poll!(map.when_insert_processed()), Poll::Pending);
460        assert_eq!(on_insert2.next().await.unwrap().into_inner(), (0, 0));
461
462        assert_eq!(poll!(map.when_insert_processed()), Poll::Ready(()));
463    }
464
465    #[tokio::test]
466    async fn on_remove_on_drop() {
467        let mut map = ProgressableHashMap::new();
468        let _ = map.insert(0, 0);
469        let _ = map.insert(1, 1);
470
471        let remove_processed = map.when_remove_processed().shared();
472        let on_remove = map.on_remove();
473
474        drop(map);
475        let removed: Vec<_> = on_remove.collect().await;
476
477        assert_eq!(poll!(remove_processed.clone()), Poll::Pending);
478        let removed: Vec<_> =
479            removed.into_iter().map(|v| v.into_inner()).collect();
480        assert_eq!(poll!(remove_processed), Poll::Ready(()));
481
482        assert_eq!(removed.len(), 2);
483        assert!(removed.contains(&(0, 0)));
484        assert!(removed.contains(&(1, 1)));
485    }
486}
medea_reactive/collections/hash_map.rs

medea_reactive/collections/
hash_map.rs
