rx_rust/operators/conditional_boolean/
take_until.rs

1use crate::utils::subscribe_with_context::{
2    self, SubscriptionContext, subscribe_with_context_owning_source,
3};
4use crate::utils::types::MaybeSend;
5use crate::{
6    disposable::Disposable,
7    observable::{Observable, Subscription},
8    observer::{Flow, Observer, Termination},
9};
10use educe::Educe;
11
12/// Emits the items emitted by a source Observable until a second Observable emits an item or a notification.
13/// See <https://reactivex.io/documentation/operators/takeuntil.html>
14///
15/// # Examples
16/// ```rust
17/// use rx_rust::{
18///     observable::ObservableExt,
19///     observer::{Observer, Termination},
20///     operators::conditional_boolean::take_until::TakeUntil,
21///     subject::publish_subject::PublishSubject,
22/// };
23/// use std::{convert::Infallible, sync::{Arc, Mutex}};
24///
25/// let values = Arc::new(Mutex::new(Vec::new()));
26/// let terminations = Arc::new(Mutex::new(Vec::new()));
27///
28/// let mut source: PublishSubject<'_, i32, Infallible> = PublishSubject::default();
29/// let mut stop: PublishSubject<'_, (), Infallible> = PublishSubject::default();
30/// let values_observer = Arc::clone(&values);
31/// let terminations_observer = Arc::clone(&terminations);
32///
33/// let subscription = TakeUntil::new(source.clone(), stop.clone()).subscribe_with_callback(
34///     move |value| values_observer.lock().unwrap().push(value),
35///     move |termination| terminations_observer
36///         .lock()
37///         .unwrap()
38///         .push(termination),
39/// );
40///
41/// source.on_next(1);
42/// source.on_next(2);
43/// stop.on_next(());
44/// source.on_next(3);
45/// drop(subscription);
46///
47/// assert_eq!(&*values.lock().unwrap(), &[1, 2]);
48/// assert_eq!(
49///     &*terminations.lock().unwrap(),
50///     &[Termination::Completed]
51/// );
52/// ```
53#[derive(Educe)]
54#[educe(Debug, Clone)]
55pub struct TakeUntil<OE, OE1> {
56    source: OE,
57    stop: OE1,
58}
59
60impl<OE, OE1> TakeUntil<OE, OE1> {
61    pub fn new<'or, T, E>(source: OE, stop: OE1) -> Self
62    where
63        OE: Observable<'or, T, E>,
64        OE1: Observable<'or, (), E>,
65    {
66        Self { source, stop }
67    }
68}
69
70impl<'or, T, E, OE, OE1> Observable<'or, T, E> for TakeUntil<OE, OE1>
71where
72    T: MaybeSend + 'or,
73    E: MaybeSend + 'or,
74    OE: Observable<'or, T, E>,
75    OE::D: MaybeSend + 'or,
76    OE1: Observable<'or, (), E>,
77    OE1::D: MaybeSend + 'or,
78{
79    type D = subscribe_with_context::OwningDisposal<'or>;
80
81    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
82        subscribe_with_context_owning_source(observer, (), |context| {
83            let subscription_1 = self.stop.subscribe(StopObserver(context.clone()));
84            let subscription_2 = self.source.subscribe(TakeUntilObserver(context));
85            subscription_1.preceded_by_bound(subscription_2)
86        })
87    }
88}
89
90struct TakeUntilObserver<T, E, OR, D: Disposable>(SubscriptionContext<T, E, OR, (), D>);
91
92impl<T, E, OR, D> Observer<T, E> for TakeUntilObserver<T, E, OR, D>
93where
94    OR: Observer<T, E>,
95    D: Disposable,
96{
97    fn on_next(&mut self, value: T) -> Flow {
98        self.0.send_next(value)
99    }
100
101    fn on_termination(self, termination: Termination<E>) {
102        self.0.send_termination(termination);
103    }
104}
105
106struct StopObserver<T, E, OR, D: Disposable>(SubscriptionContext<T, E, OR, (), D>);
107
108impl<T, E, OR, D> Observer<(), E> for StopObserver<T, E, OR, D>
109where
110    OR: Observer<T, E>,
111    D: Disposable,
112{
113    fn on_next(&mut self, _: ()) -> Flow {
114        self.0.send_termination(Termination::Completed);
115        // The first notification ends the stream, so the notifier is of no use afterwards.
116        Flow::Stop
117    }
118
119    fn on_termination(self, termination: Termination<E>) {
120        self.0.send_termination(termination);
121    }
122}
rx_rust/operators/conditional_boolean/take_until.rs

rx_rust/operators/conditional_boolean/
take_until.rs
