rx_rust/operators/conditional_boolean/
take_while.rs

1use crate::utils::types::NecessarySendSync;
2use crate::utils::unsub_after_termination::subscribe_unsub_after_termination;
3use crate::{
4    disposable::subscription::Subscription,
5    observable::Observable,
6    observer::{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::observable_ext::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, 'sub, T, E>(source: OE, callback: F) -> Self
45    where
46        OE: Observable<'or, 'sub, T, E>,
47        F: FnMut(&T) -> bool,
48    {
49        Self { source, callback }
50    }
51}
52
53impl<'or, 'sub, T, E, OE, F> Observable<'or, 'sub, T, E> for TakeWhile<OE, F>
54where
55    OE: Observable<'or, 'sub, T, E>,
56    F: FnMut(&T) -> bool + NecessarySendSync + 'or,
57    'sub: 'or,
58{
59    fn subscribe(
60        self,
61        observer: impl Observer<T, E> + NecessarySendSync + 'or,
62    ) -> Subscription<'sub> {
63        subscribe_unsub_after_termination(observer, |observer| {
64            let observer = TakeWhileObserver {
65                observer: Some(observer),
66                callback: self.callback,
67            };
68            self.source.subscribe(observer)
69        })
70    }
71}
72
73struct TakeWhileObserver<OR, F> {
74    observer: Option<OR>,
75    callback: F,
76}
77
78impl<T, E, OR, F> Observer<T, E> for TakeWhileObserver<OR, F>
79where
80    OR: Observer<T, E>,
81    F: FnMut(&T) -> bool,
82{
83    fn on_next(&mut self, value: T) {
84        let r#continue = (self.callback)(&value);
85        if r#continue {
86            if let Some(observer) = self.observer.as_mut() {
87                observer.on_next(value);
88            }
89        } else if let Some(observer) = self.observer.take() {
90            observer.on_termination(Termination::Completed);
91        }
92    }
93
94    fn on_termination(self, termination: Termination<E>) {
95        if let Some(observer) = self.observer {
96            observer.on_termination(termination);
97        }
98    }
99}
rx_rust/operators/conditional_boolean/take_while.rs

rx_rust/operators/conditional_boolean/
take_while.rs
