rx_rust/operators/mathematical_aggregate/
sum.rs

1use crate::utils::types::NecessarySend;
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 + NecessarySend + 'or,
54    OE: Observable<'or, 'sub, T, E>,
55{
56    fn subscribe(self, observer: impl Observer<T, E> + NecessarySend + 'or) -> Subscription<'sub> {
57        let observer = SumObserver {
58            observer,
59            sum: None,
60        };
61        self.source.subscribe(observer)
62    }
63}
64
65struct SumObserver<T, OR> {
66    observer: OR,
67    sum: Option<T>,
68}
69
70impl<T, E, OR> Observer<T, E> for SumObserver<T, OR>
71where
72    T: AddAssign,
73    OR: Observer<T, E>,
74{
75    fn on_next(&mut self, value: T) {
76        if let Some(sum) = &mut self.sum {
77            *sum += value;
78        } else {
79            self.sum = Some(value);
80        }
81    }
82
83    fn on_termination(mut self, termination: Termination<E>) {
84        match termination {
85            Termination::Completed => {
86                if let Some(sum) = self.sum {
87                    self.observer.on_next(sum);
88                }
89            }
90            Termination::Error(_) => {}
91        }
92        self.observer.on_termination(termination)
93    }
94}
rx_rust/operators/mathematical_aggregate/sum.rs

rx_rust/operators/mathematical_aggregate/
sum.rs
