rx_rust/operators/mathematical_aggregate/
collect.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    observable::Observable,
4    observable::Subscription,
5    observer::{Flow, Observer, Termination},
6    utils::types::MarkerType,
7};
8use educe::Educe;
9use std::marker::PhantomData;
10
11/// Gathers all the items emitted by an Observable into a single collection and emits it when the
12/// source completes.
13///
14/// The collection is built with [`Default`] and [`Extend`], so it can be a `Vec<T>`, a
15/// `HashSet<T>`, a `String` of `char`s, or any other type that implements both.
16/// [`to_vec`](crate::observable::ObservableExt::to_vec) is the `Vec<T>` case.
17///
18/// A source that completes without emitting yields the empty collection, an error discards the
19/// items gathered so far and is forwarded on its own, and a source that never terminates never
20/// emits while gathering its items without bound.
21/// See <https://reactivex.io/documentation/operators/to.html>
22///
23/// # Examples
24/// ```rust
25/// use rx_rust::{
26///     observable::ObservableExt,
27///     observer::Termination,
28///     operators::creating::from_iter::FromIter,
29/// };
30/// use std::collections::BTreeSet;
31///
32/// let mut values = Vec::new();
33/// let mut terminations = Vec::new();
34///
35/// let observable = FromIter::new(vec![1, 2, 2, 3]).collect::<BTreeSet<_>>();
36/// observable.subscribe_with_callback(
37///     |value| values.push(value),
38///     |termination| terminations.push(termination),
39/// );
40///
41/// assert_eq!(values, vec![BTreeSet::from([1, 2, 3])]);
42/// assert_eq!(terminations, vec![Termination::Completed]);
43/// ```
44#[derive(Educe)]
45#[educe(Debug, Clone)]
46pub struct Collect<C, T, OE> {
47    source: OE,
48    _marker: MarkerType<(C, T)>,
49}
50
51impl<C, T, OE> Collect<C, T, OE> {
52    pub fn new<'or, E>(source: OE) -> Self
53    where
54        OE: Observable<'or, T, E>,
55        C: Default + Extend<T>,
56    {
57        Self {
58            source,
59            _marker: PhantomData,
60        }
61    }
62}
63
64impl<'or, C, T, E, OE> Observable<'or, C, E> for Collect<C, T, OE>
65where
66    C: Default + Extend<T> + MaybeSend + 'or,
67    OE: Observable<'or, T, E>,
68{
69    type D = OE::D;
70
71    fn subscribe(self, observer: impl Observer<C, E> + MaybeSend + 'or) -> Subscription<Self::D> {
72        let observer = CollectObserver {
73            observer,
74            collection: C::default(),
75        };
76        self.source.subscribe(observer)
77    }
78}
79
80struct CollectObserver<C, OR> {
81    observer: OR,
82    collection: C,
83}
84
85impl<C, T, E, OR> Observer<T, E> for CollectObserver<C, OR>
86where
87    C: Extend<T>,
88    OR: Observer<C, E>,
89{
90    fn on_next(&mut self, value: T) -> Flow {
91        self.collection.extend(Some(value));
92        Flow::Continue
93    }
94
95    fn on_termination(mut self, termination: Termination<E>) {
96        // The final value ends the stream, so a downstream that stopped on it is not completed
97        // on top of that: it has already ended itself.
98        if matches!(termination, Termination::Completed)
99            && self.observer.on_next(self.collection).is_stop()
100        {
101            return;
102        }
103        self.observer.on_termination(termination)
104    }
105}
rx_rust/operators/mathematical_aggregate/collect.rs

rx_rust/operators/mathematical_aggregate/
collect.rs
