rx_rust/operators/mathematical_aggregate/
sum.rs

1use crate::utils::types::NecessarySendSync;
2use crate::{
3    disposable::subscription::Subscription,
4    observable::Observable,
5    observer::{Observer, Termination},
6};
7use educe::Educe;
8use std::ops::AddAssign;
9
10/// Calculates the sum of numbers emitted by an Observable and emits this sum.
11/// See <https://reactivex.io/documentation/operators/sum.html>
12///
13/// # Examples
14/// ```rust
15/// use rx_rust::{
16///     observable::observable_ext::ObservableExt,
17///     observer::Termination,
18///     operators::{
19///         creating::from_iter::FromIter,
20///         mathematical_aggregate::sum::Sum,
21///     },
22/// };
23///
24/// let mut values = Vec::new();
25/// let mut terminations = Vec::new();
26///
27/// let observable = Sum::new(FromIter::new(vec![1, 2, 3]));
28/// observable.subscribe_with_callback(
29///     |value| values.push(value),
30///     |termination| terminations.push(termination),
31/// );
32///
33/// assert_eq!(values, vec![6]);
34/// assert_eq!(terminations, vec![Termination::Completed]);
35/// ```
36#[derive(Educe)]
37#[educe(Debug, Clone)]
38pub struct Sum<OE> {
39    source: OE,
40}
41
42impl<OE> Sum<OE> {
43    pub fn new<'or, 'sub, T, E>(source: OE) -> Self
44    where
45        OE: Observable<'or, 'sub, T, E>,
46    {
47        Self { source }
48    }
49}
50
51impl<'or, 'sub, T, E, OE> Observable<'or, 'sub, T, E> for Sum<OE>
52where
53    T: AddAssign + NecessarySendSync + 'or,
54    OE: Observable<'or, 'sub, T, E>,
55{
56    fn subscribe(
57        self,
58        observer: impl Observer<T, E> + NecessarySendSync + 'or,
59    ) -> Subscription<'sub> {
60        let observer = SumObserver {
61            observer,
62            sum: None,
63        };
64        self.source.subscribe(observer)
65    }
66}
67
68struct SumObserver<T, OR> {
69    observer: OR,
70    sum: Option<T>,
71}
72
73impl<T, E, OR> Observer<T, E> for SumObserver<T, OR>
74where
75    T: AddAssign,
76    OR: Observer<T, E>,
77{
78    fn on_next(&mut self, value: T) {
79        if let Some(sum) = &mut self.sum {
80            *sum += value;
81        } else {
82            self.sum = Some(value);
83        }
84    }
85
86    fn on_termination(mut self, termination: Termination<E>) {
87        match termination {
88            Termination::Completed => {
89                if let Some(sum) = self.sum {
90                    self.observer.on_next(sum);
91                }
92            }
93            Termination::Error(_) => {}
94        }
95        self.observer.on_termination(termination)
96    }
97}
rx_rust/operators/mathematical_aggregate/sum.rs

rx_rust/operators/mathematical_aggregate/
sum.rs
