rx_rust/operators/mathematical_aggregate/
max.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    observable::Observable,
4    observable::Subscription,
5    observer::{Flow, Observer, Termination},
6};
7use educe::Educe;
8
9/// Emits the maximum item emitted by an Observable.
10/// See <https://reactivex.io/documentation/operators/max.html>
11///
12/// `T` is only [`PartialOrd`], so values that do not compare — `f64::NAN` among them — are
13/// never seen as greater and are skipped. A `NaN` that arrives first is therefore kept as the
14/// maximum for the rest of the stream, because nothing compares greater than it.
15///
16/// # Examples
17/// ```rust
18/// use rx_rust::{
19///     observable::ObservableExt,
20///     observer::Termination,
21///     operators::{
22///         creating::from_iter::FromIter,
23///         mathematical_aggregate::max::Max,
24///     },
25/// };
26///
27/// let mut values = Vec::new();
28/// let mut terminations = Vec::new();
29///
30/// let observable = Max::new(FromIter::new(vec![3, 5, 4]));
31/// observable.subscribe_with_callback(
32///     |value| values.push(value),
33///     |termination| terminations.push(termination),
34/// );
35///
36/// assert_eq!(values, vec![5]);
37/// assert_eq!(terminations, vec![Termination::Completed]);
38/// ```
39#[derive(Educe)]
40#[educe(Debug, Clone)]
41pub struct Max<OE> {
42    source: OE,
43}
44
45impl<OE> Max<OE> {
46    pub fn new<'or, T, E>(source: OE) -> Self
47    where
48        OE: Observable<'or, T, E>,
49    {
50        Self { source }
51    }
52}
53
54impl<'or, T, E, OE> Observable<'or, T, E> for Max<OE>
55where
56    T: PartialOrd + MaybeSend + 'or,
57    OE: Observable<'or, T, E>,
58{
59    type D = OE::D;
60
61    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
62        let observer = MaxObserver {
63            observer,
64            max: None,
65        };
66        self.source.subscribe(observer)
67    }
68}
69
70struct MaxObserver<T, OR> {
71    observer: OR,
72    max: Option<T>,
73}
74
75impl<T, E, OR> Observer<T, E> for MaxObserver<T, OR>
76where
77    T: PartialOrd,
78    OR: Observer<T, E>,
79{
80    fn on_next(&mut self, value: T) -> Flow {
81        if let Some(max) = &mut self.max {
82            if value > *max {
83                *max = value;
84            }
85        } else {
86            self.max = Some(value);
87        }
88        Flow::Continue
89    }
90
91    fn on_termination(mut self, termination: Termination<E>) {
92        // The final value ends the stream, so a downstream that stopped on it is not completed
93        // on top of that: it has already ended itself.
94        if matches!(termination, Termination::Completed)
95            && let Some(max) = self.max.take()
96            && self.observer.on_next(max).is_stop()
97        {
98            return;
99        }
100        self.observer.on_termination(termination)
101    }
102}
rx_rust/operators/mathematical_aggregate/max.rs

rx_rust/operators/mathematical_aggregate/
max.rs
