rx_rust/operators/mathematical_aggregate/
average.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/// Calculates the average of numbers emitted by an Observable and emits this average.
11/// See <https://reactivex.io/documentation/operators/average.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::average::Average,
21///     },
22/// };
23///
24/// let mut values = Vec::new();
25/// let mut terminations = Vec::new();
26///
27/// let observable = Average::new(FromIter::new(vec![1.0_f64, 3.0, 5.0]));
28/// observable.subscribe_with_callback(
29///     |value| values.push(value),
30///     |termination| terminations.push(termination),
31/// );
32///
33/// assert_eq!(values, vec![3.0]);
34/// assert_eq!(terminations, vec![Termination::Completed]);
35/// ```
36#[derive(Educe)]
37#[educe(Debug, Clone)]
38pub struct Average<T, OE> {
39    source: OE,
40    _marker: MarkerType<T>,
41}
42
43impl<T, OE> Average<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
55struct AverageObserver<T, OR> {
56    observer: OR,
57    /// Accumulated in `f64`, the type of the result. Summing in the source's own type overflows
58    /// for the narrow ones long before the stream ends: three `100u8` items already exceed
59    /// `u8::MAX`.
60    sum: f64,
61    count: usize,
62    _marker: MarkerType<T>,
63}
64
65macro_rules! average_observer_impl {
66    ($($t:ty)*) => ($(
67
68        impl<'or, E, OE> Observable<'or, f64, E> for Average<$t, OE>
69        where
70            OE: Observable<'or, $t, E>,
71        {
72            type D = OE::D;
73
74            fn subscribe(self, observer: impl Observer<f64, E> + MaybeSend + 'or) -> Subscription<Self::D> {
75                let observer = AverageObserver {
76                    observer,
77                    sum: 0f64,
78                    count: 0,
79                    _marker: PhantomData,
80                };
81                self.source.subscribe(observer)
82            }
83        }
84
85        impl<E, OR> Observer<$t, E> for AverageObserver<$t, OR>
86        where
87            OR: Observer<f64, E>,
88        {
89            fn on_next(&mut self, value: $t) -> Flow {
90                self.sum += value as f64;
91                self.count += 1;
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
97                // completed on top of that: it has already ended itself.
98                if matches!(termination, Termination::Completed)
99                    && self.count != 0
100                    && self.observer.on_next(self.sum / self.count as f64).is_stop()
101                {
102                    return;
103                }
104                self.observer.on_termination(termination)
105            }
106        }
107
108    )*)
109}
110
111average_observer_impl! { usize u8 u16 u32 u64 u128 isize i8 i16 i32 i64 i128 f32 f64 }
rx_rust/operators/mathematical_aggregate/average.rs

rx_rust/operators/mathematical_aggregate/
average.rs
