rx_rust/operators/combining/
switch.rs

1use crate::disposable::Disposable;
2use crate::operators::others::with_error_type::WithErrorType;
3use crate::utils::id_generator::{Id, IdGenerator};
4use crate::utils::serialized_delivery::UpdateOutcome;
5use crate::utils::subscribe_with_context::{
6    self, SubscriptionContext, subscribe_with_context_owning_source,
7};
8use crate::utils::subscription_slot::SubscriptionSlot;
9use crate::utils::types::MaybeSend;
10use crate::{
11    observable::{Observable, Subscription},
12    observer::{Flow, Observer, Termination},
13    operators::creating::from_iter::FromIter,
14    utils::types::MarkerType,
15};
16use educe::Educe;
17use std::marker::PhantomData;
18
19/// Converts an Observable that emits Observables into a single Observable that emits the items emitted by the most recently emitted of those Observables.
20/// See <https://reactivex.io/documentation/operators/switch.html>
21///
22/// # Examples
23/// ```rust
24/// use rx_rust::{
25///     observable::ObservableExt,
26///     observer::Termination,
27///     operators::{
28///         combining::switch::Switch,
29///         creating::from_iter::FromIter,
30///     },
31/// };
32///
33/// let mut values = Vec::new();
34/// let mut terminations = Vec::new();
35///
36/// let inner_1 = FromIter::new(vec![1, 2]);
37/// let inner_2 = FromIter::new(vec![3, 4]);
38/// let observable = Switch::new_from_iter([inner_1, inner_2]);
39/// observable.subscribe_with_callback(
40///     |value| values.push(value),
41///     |termination| terminations.push(termination),
42/// );
43///
44/// assert_eq!(values, vec![1, 2, 3, 4]);
45/// assert_eq!(terminations, vec![Termination::Completed]);
46/// ```
47#[derive(Educe)]
48#[educe(Debug, Clone)]
49pub struct Switch<OE, OE1> {
50    source: OE,
51    _marker: MarkerType<OE1>,
52}
53
54impl<OE, OE1> Switch<OE, OE1> {
55    pub fn new<'or, T, E>(source: OE) -> Self
56    where
57        OE: Observable<'or, OE1, E>,
58        OE1: Observable<'or, T, E>,
59    {
60        Self {
61            source,
62            _marker: PhantomData,
63        }
64    }
65}
66
67impl<E, OE1, I> Switch<WithErrorType<E, FromIter<I>>, OE1> {
68    pub fn new_from_iter<'or, T>(into_iterator: I) -> Self
69    where
70        I: IntoIterator<Item = OE1>,
71        OE1: Observable<'or, T, E>,
72    {
73        Self {
74            source: WithErrorType::new(FromIter::new(into_iterator)),
75            _marker: PhantomData,
76        }
77    }
78}
79
80impl<'or, T, E, OE, OE1> Observable<'or, T, E> for Switch<OE, OE1>
81where
82    T: MaybeSend + 'or,
83    E: MaybeSend + 'or,
84    OE: Observable<'or, OE1, E>,
85    OE::D: MaybeSend + 'or,
86    OE1: Observable<'or, T, E>,
87    OE1::D: MaybeSend + 'or,
88{
89    type D = subscribe_with_context::OwningDisposal<'or>;
90
91    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
92        let model = Model {
93            slot: SubscriptionSlot::Idle,
94            is_source_completed: false,
95            sub_ids: IdGenerator::default(),
96        };
97        subscribe_with_context_owning_source(observer, model, |context| {
98            self.source.subscribe(SwitchObserver(context))
99        })
100    }
101}
102
103struct Model<D: Disposable> {
104    slot: SubscriptionSlot<Subscription<D>>,
105    is_source_completed: bool,
106    /// The current inner subscription is always the one subscribed last, so the id it was handed
107    /// is [`IdGenerator::latest`]. An inner observer whose id is no longer the latest was
108    /// superseded, and its events are dropped: the subscription it replaced is disposed outside
109    /// the lock, so it can still deliver one after being replaced.
110    sub_ids: IdGenerator,
111}
112
113struct SwitchObserver<T, E, OR, ID: Disposable, SD: Disposable>(
114    SubscriptionContext<T, E, OR, Model<ID>, SD>,
115);
116
117impl<'or, T, E, OR, OE1, SD> Observer<OE1, E> for SwitchObserver<T, E, OR, OE1::D, SD>
118where
119    T: MaybeSend + 'or,
120    E: MaybeSend + 'or,
121    OR: Observer<T, E> + MaybeSend + 'or,
122    OE1: Observable<'or, T, E>,
123    OE1::D: MaybeSend + 'or,
124    SD: Disposable + MaybeSend + 'or,
125{
126    fn on_next(&mut self, value: OE1) -> Flow {
127        let result = self.0.update(|model| {
128            UpdateOutcome::new(model.sub_ids.next_id()).with_drop_outside(model.slot.reserve())
129        });
130        let sub_id = match result {
131            Ok(sub_id) => sub_id,
132            Err(_) => return Flow::Stop,
133        };
134        let observer = SwitchInnerObserver(self.0.clone(), sub_id);
135        let sub = value.subscribe(observer);
136        // `fill` gives the subscription back when the slot was released while it was being built,
137        // which means the operator already terminated.
138        self.0
139            .update_flow(|model| UpdateOutcome::empty().with_drop_outside(model.slot.fill(sub)))
140    }
141
142    fn on_termination(self, termination: Termination<E>) {
143        match termination {
144            completion @ Termination::Completed => {
145                let _ = self.0.update(|model| {
146                    model.is_source_completed = true;
147                    if model.slot.is_idle() {
148                        UpdateOutcome::empty().with_termination_event(completion)
149                    } else {
150                        UpdateOutcome::empty().without_events()
151                    }
152                });
153            }
154            error @ Termination::Error(_) => {
155                self.0.send_termination(error);
156            }
157        }
158    }
159}
160
161struct SwitchInnerObserver<T, E, OR, ID: Disposable, SD: Disposable>(
162    SubscriptionContext<T, E, OR, Model<ID>, SD>,
163    Id,
164);
165
166impl<T, E, OR, ID, SD> Observer<T, E> for SwitchInnerObserver<T, E, OR, ID, SD>
167where
168    OR: Observer<T, E>,
169    ID: Disposable,
170    SD: Disposable,
171{
172    fn on_next(&mut self, value: T) -> Flow {
173        self.0.update_flow(|model| {
174            if model.sub_ids.latest() != Some(self.1) {
175                return UpdateOutcome::empty().without_events();
176            }
177            UpdateOutcome::empty().with_next_event(value)
178        })
179    }
180
181    fn on_termination(self, termination: Termination<E>) {
182        let _ = self.0.update(|model| {
183            if model.sub_ids.latest() != Some(self.1) {
184                return UpdateOutcome::empty()
185                    .without_events()
186                    .without_drop_outside();
187            }
188            match termination {
189                completion @ Termination::Completed => {
190                    if model.is_source_completed {
191                        UpdateOutcome::empty()
192                            .with_termination_event(completion)
193                            .without_drop_outside()
194                    } else {
195                        assert!(
196                            !model.slot.is_idle(),
197                            "the terminating inner subscription is still held or reserved"
198                        );
199                        UpdateOutcome::empty()
200                            .without_events()
201                            .with_drop_outside(model.slot.release())
202                    }
203                }
204                error @ Termination::Error(_) => UpdateOutcome::empty()
205                    .with_termination_event(error)
206                    .without_drop_outside(),
207            }
208        });
209    }
210}
rx_rust/operators/combining/switch.rs

rx_rust/operators/combining/
switch.rs
