rx_rust/operators/conditional_boolean/
take_while.rs

1use crate::utils::types::NecessarySend;
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 + NecessarySend + 'or,
57    'sub: 'or,
58{
59    fn subscribe(self, observer: impl Observer<T, E> + NecessarySend + 'or) -> Subscription<'sub> {
60        subscribe_unsub_after_termination(observer, |observer| {
61            let observer = TakeWhileObserver {
62                observer: Some(observer),
63                callback: self.callback,
64            };
65            self.source.subscribe(observer)
66        })
67    }
68}
69
70struct TakeWhileObserver<OR, F> {
71    observer: Option<OR>,
72    callback: F,
73}
74
75impl<T, E, OR, F> Observer<T, E> for TakeWhileObserver<OR, F>
76where
77    OR: Observer<T, E>,
78    F: FnMut(&T) -> bool,
79{
80    fn on_next(&mut self, value: T) {
81        let r#continue = (self.callback)(&value);
82        if r#continue {
83            if let Some(observer) = self.observer.as_mut() {
84                observer.on_next(value);
85            }
86        } else if let Some(observer) = self.observer.take() {
87            observer.on_termination(Termination::Completed);
88        }
89    }
90
91    fn on_termination(self, termination: Termination<E>) {
92        if let Some(observer) = self.observer {
93            observer.on_termination(termination);
94        }
95    }
96}
rx_rust/operators/conditional_boolean/take_while.rs

rx_rust/operators/conditional_boolean/
take_while.rs
