rx_rust/operators/conditional_boolean/
all.rs

1use crate::observable::Subscription;
2use crate::utils::subscribe_with_auto_dispose_on_termination::{
3    self, subscribe_with_auto_dispose_on_termination,
4};
5use crate::utils::types::{MarkerType, MaybeSend};
6use crate::{
7    observable::Observable,
8    observer::{Flow, Observer, Termination},
9};
10use educe::Educe;
11use std::marker::PhantomData;
12
13/// Emits a single boolean value that indicates whether all items emitted by a source Observable satisfy a specified condition.
14/// See <https://reactivex.io/documentation/operators/all.html>
15///
16/// # Examples
17/// ```rust
18/// use rx_rust::{
19///     observable::ObservableExt,
20///     observer::Termination,
21///     operators::{
22///         conditional_boolean::all::All,
23///         creating::from_iter::FromIter,
24///     },
25/// };
26///
27/// let mut values = Vec::new();
28/// let mut terminations = Vec::new();
29///
30/// let observable = All::new(FromIter::new(vec![1, 2, 3]), |value| value < 5);
31/// observable.subscribe_with_callback(
32///     |value| values.push(value),
33///     |termination| terminations.push(termination),
34/// );
35///
36/// assert_eq!(values, vec![true]);
37/// assert_eq!(terminations, vec![Termination::Completed]);
38/// ```
39#[derive(Educe)]
40#[educe(Debug, Clone)]
41pub struct All<T, OE, F> {
42    source: OE,
43    callback: F,
44    _marker: MarkerType<T>,
45}
46
47impl<T, OE, F> All<T, OE, F> {
48    pub fn new<'or, E>(source: OE, callback: F) -> Self
49    where
50        OE: Observable<'or, T, E>,
51        F: FnMut(T) -> bool,
52    {
53        Self {
54            source,
55            callback,
56            _marker: PhantomData,
57        }
58    }
59}
60
61impl<'or, T, E, OE, F> Observable<'or, bool, E> for All<T, OE, F>
62where
63    OE: Observable<'or, T, E>,
64    OE::D: MaybeSend + 'or,
65    F: FnMut(T) -> bool + MaybeSend + 'or,
66{
67    type D = subscribe_with_auto_dispose_on_termination::Disposal<OE::D>;
68
69    fn subscribe(
70        self,
71        observer: impl Observer<bool, E> + MaybeSend + 'or,
72    ) -> Subscription<Self::D> {
73        subscribe_with_auto_dispose_on_termination(observer, |observer| {
74            let observer = AllObserver {
75                observer: Some(observer),
76                callback: self.callback,
77            };
78            self.source.subscribe(observer)
79        })
80    }
81}
82
83struct AllObserver<OR, F> {
84    observer: Option<OR>,
85    callback: F,
86}
87
88impl<T, E, OR, F> Observer<T, E> for AllObserver<OR, F>
89where
90    OR: Observer<bool, E>,
91    F: FnMut(T) -> bool,
92{
93    fn on_next(&mut self, value: T) -> Flow {
94        // A source that does not honor the flow or the disposal keeps emitting; the callback is
95        // the caller's and may have side effects, so it must not run once the result was decided.
96        if self.observer.is_none() {
97            return Flow::Stop;
98        }
99        if (self.callback)(value) {
100            return Flow::Continue;
101        }
102        // One value that fails decides the result, so the rest of the source is of no use.
103        let Some(mut observer) = self.observer.take() else {
104            return Flow::Stop;
105        };
106        if observer.on_next(false).is_continue() {
107            observer.on_termination(Termination::Completed);
108        }
109        Flow::Stop
110    }
111
112    fn on_termination(mut self, termination: Termination<E>) {
113        if let Some(mut observer) = self.observer.take() {
114            match termination {
115                Termination::Completed => drop(observer.on_next(true)),
116                Termination::Error(_) => {}
117            }
118            observer.on_termination(termination);
119        }
120    }
121}
rx_rust/operators/conditional_boolean/all.rs

rx_rust/operators/conditional_boolean/
all.rs
