rx_rust/operators/conditional_boolean/
take_while.rs

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

rx_rust/operators/conditional_boolean/
take_while.rs
