rx_rust/operators/mathematical_aggregate/
count.rs

1use crate::utils::types::{MarkerType, MaybeSend};
2use crate::{
3    observable::Observable,
4    observable::Subscription,
5    observer::{Flow, Observer, Termination},
6};
7use educe::Educe;
8use std::marker::PhantomData;
9
10/// Counts the number of items emitted by the source Observable and emits this count.
11/// See <https://reactivex.io/documentation/operators/count.html>
12///
13/// # Examples
14/// ```rust
15/// use rx_rust::{
16///     observable::ObservableExt,
17///     observer::Termination,
18///     operators::{
19///         creating::from_iter::FromIter,
20///         mathematical_aggregate::count::Count,
21///     },
22/// };
23///
24/// let mut values = Vec::new();
25/// let mut terminations = Vec::new();
26///
27/// let observable = Count::new(FromIter::new(vec![1, 2, 3, 4]));
28/// observable.subscribe_with_callback(
29///     |value| values.push(value),
30///     |termination| terminations.push(termination),
31/// );
32///
33/// assert_eq!(values, vec![4]);
34/// assert_eq!(terminations, vec![Termination::Completed]);
35/// ```
36#[derive(Educe)]
37#[educe(Debug, Clone)]
38pub struct Count<T, OE> {
39    source: OE,
40    _marker: MarkerType<T>,
41}
42
43impl<T, OE> Count<T, OE> {
44    pub fn new<'or, E>(source: OE) -> Self
45    where
46        OE: Observable<'or, T, E>,
47    {
48        Self {
49            source,
50            _marker: PhantomData,
51        }
52    }
53}
54
55impl<'or, T, E, OE> Observable<'or, usize, E> for Count<T, OE>
56where
57    OE: Observable<'or, T, E>,
58{
59    type D = OE::D;
60
61    fn subscribe(
62        self,
63        observer: impl Observer<usize, E> + MaybeSend + 'or,
64    ) -> Subscription<Self::D> {
65        let observer = CountObserver { observer, count: 0 };
66        self.source.subscribe(observer)
67    }
68}
69
70struct CountObserver<OR> {
71    observer: OR,
72    count: usize,
73}
74
75impl<T, E, OR> Observer<T, E> for CountObserver<OR>
76where
77    OR: Observer<usize, E>,
78{
79    fn on_next(&mut self, _: T) -> Flow {
80        self.count += 1;
81        Flow::Continue
82    }
83
84    fn on_termination(mut self, termination: Termination<E>) {
85        // The final value ends the stream, so a downstream that stopped on it is not completed
86        // on top of that: it has already ended itself.
87        if matches!(termination, Termination::Completed)
88            && self.observer.on_next(self.count).is_stop()
89        {
90            return;
91        }
92        self.observer.on_termination(termination)
93    }
94}
rx_rust/operators/mathematical_aggregate/count.rs

rx_rust/operators/mathematical_aggregate/
count.rs
