rx_rust/operators/conditional_boolean/
skip_until.rs

1use crate::disposable::Disposable;
2use crate::utils::serialized_delivery::UpdateOutcome;
3use crate::utils::subscribe_with_context::{
4    self, SubscriptionContext, subscribe_with_context_owning_source,
5};
6use crate::utils::types::MaybeSend;
7use crate::{
8    observable::Observable,
9    observable::Subscription,
10    observer::{Flow, Observer, Termination},
11};
12use educe::Educe;
13
14/// Discards items emitted by a source Observable until a second Observable emits an item.
15/// See <https://reactivex.io/documentation/operators/skipuntil.html>
16///
17/// # Examples
18/// ```rust
19/// use rx_rust::{
20///     observable::ObservableExt,
21///     observer::{Observer, Termination},
22///     operators::conditional_boolean::skip_until::SkipUntil,
23///     subject::publish_subject::PublishSubject,
24/// };
25/// use std::{convert::Infallible, sync::{Arc, Mutex}};
26///
27/// let values = Arc::new(Mutex::new(Vec::new()));
28/// let terminations = Arc::new(Mutex::new(Vec::new()));
29///
30/// let mut source: PublishSubject<'_, i32, Infallible> = PublishSubject::default();
31/// let mut gate: PublishSubject<'_, (), Infallible> = PublishSubject::default();
32/// let values_observer = Arc::clone(&values);
33/// let terminations_observer = Arc::clone(&terminations);
34///
35/// let subscription = SkipUntil::new(source.clone(), gate.clone()).subscribe_with_callback(
36///     move |value| values_observer.lock().unwrap().push(value),
37///     move |termination| terminations_observer
38///         .lock()
39///         .unwrap()
40///         .push(termination),
41/// );
42///
43/// source.on_next(1);
44/// gate.on_next(());
45/// source.on_next(2);
46/// source.on_termination(Termination::Completed);
47/// drop(subscription);
48///
49/// assert_eq!(&*values.lock().unwrap(), &[2]);
50/// assert_eq!(
51///     &*terminations.lock().unwrap(),
52///     &[Termination::Completed]
53/// );
54/// ```
55#[derive(Educe)]
56#[educe(Debug, Clone)]
57pub struct SkipUntil<OE, OE1> {
58    source: OE,
59    start: OE1,
60}
61
62impl<OE, OE1> SkipUntil<OE, OE1> {
63    pub fn new<'or, T, E>(source: OE, start: OE1) -> Self
64    where
65        OE: Observable<'or, T, E>,
66        OE1: Observable<'or, (), E>,
67    {
68        Self { source, start }
69    }
70}
71
72impl<'or, T, E, OE, OE1> Observable<'or, T, E> for SkipUntil<OE, OE1>
73where
74    T: MaybeSend + 'or,
75    E: MaybeSend + 'or,
76    OE: Observable<'or, T, E>,
77    OE::D: MaybeSend + 'or,
78    OE1: Observable<'or, (), E>,
79    OE1::D: MaybeSend + 'or,
80{
81    type D = subscribe_with_context::OwningDisposal<'or>;
82
83    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
84        let model = Model { started: false };
85        subscribe_with_context_owning_source(observer, model, |context| {
86            let subscription_1 = self.start.subscribe(StartObserver {
87                context: context.clone(),
88                started: false,
89            });
90            let subscription_2 = self.source.subscribe(SkipUntilObserver(context));
91            subscription_1.preceded_by_bound(subscription_2)
92        })
93    }
94}
95
96struct Model {
97    started: bool,
98}
99
100struct SkipUntilObserver<T, E, OR, D: Disposable>(SubscriptionContext<T, E, OR, Model, D>);
101
102impl<T, E, OR, D> Observer<T, E> for SkipUntilObserver<T, E, OR, D>
103where
104    OR: Observer<T, E>,
105    D: Disposable,
106{
107    fn on_next(&mut self, value: T) -> Flow {
108        self.0.update_flow(|model| {
109            if model.started {
110                UpdateOutcome::empty().with_next_event(value)
111            } else {
112                UpdateOutcome::empty().without_events()
113            }
114        })
115    }
116
117    fn on_termination(self, termination: Termination<E>) {
118        self.0.send_termination(termination);
119    }
120}
121
122struct StartObserver<T, E, OR, D: Disposable> {
123    context: SubscriptionContext<T, E, OR, Model, D>,
124    started: bool,
125}
126
127impl<T, E, OR, D> Observer<(), E> for StartObserver<T, E, OR, D>
128where
129    OR: Observer<T, E>,
130    D: Disposable,
131{
132    fn on_next(&mut self, _: ()) -> Flow {
133        if !self.started {
134            self.started = true;
135            self.context.update_flow(|model| {
136                model.started = true;
137                UpdateOutcome::empty()
138            })
139        } else {
140            Flow::Continue
141        }
142    }
143
144    fn on_termination(self, termination: Termination<E>) {
145        match termination {
146            completion @ Termination::Completed => {
147                if !self.started {
148                    self.context.send_termination(completion);
149                }
150            }
151            error @ Termination::Error(_) => {
152                self.context.send_termination(error);
153            }
154        }
155    }
156}
rx_rust/operators/conditional_boolean/skip_until.rs

rx_rust/operators/conditional_boolean/
skip_until.rs
