rx_rust/operators/combining/
concat_all.rs

1use crate::disposable::Disposable;
2use crate::operators::others::with_error_type::WithErrorType;
3use crate::utils::serialized_delivery::{DeliveryStopped, UpdateOutcome};
4use crate::utils::subscribe_with_context::{
5    self, SubscriptionContext, subscribe_with_context_owning_source,
6};
7use crate::utils::subscription_slot::SubscriptionSlot;
8use crate::utils::types::MaybeSend;
9use crate::{
10    observable::{Observable, Subscription},
11    observer::{Flow, Observer, Termination},
12    operators::creating::from_iter::FromIter,
13    utils::types::MarkerType,
14};
15use educe::Educe;
16use std::{collections::VecDeque, marker::PhantomData};
17
18/// Concatenates an Observable of Observables, emitting all values from each inner Observable in sequence.
19/// See <https://reactivex.io/documentation/operators/concat.html> (referencing concat operator for general concept)
20///
21/// # Examples
22/// ```rust
23/// use rx_rust::{
24///     observable::ObservableExt,
25///     observer::Termination,
26///     operators::{
27///         combining::concat_all::ConcatAll,
28///         creating::from_iter::FromIter,
29///     },
30/// };
31///
32/// let mut values = Vec::new();
33/// let mut terminations = Vec::new();
34///
35/// let observable = ConcatAll::new_from_iter([
36///     FromIter::new(vec![1, 2]),
37///     FromIter::new(vec![3, 4]),
38/// ]);
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 ConcatAll<OE, OE1> {
50    source: OE,
51    _marker: MarkerType<OE1>,
52}
53
54impl<OE, OE1> ConcatAll<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> ConcatAll<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 ConcatAll<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> + MaybeSend + 'or,
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            pending_observables: VecDeque::new(),
94            slot: SubscriptionSlot::Idle,
95            is_source_completed: false,
96        };
97        subscribe_with_context_owning_source(observer, model, |context| {
98            self.source.subscribe(SourceObserver(context.clone()))
99        })
100    }
101}
102
103struct Model<'or, T, E, OE1>
104where
105    OE1: Observable<'or, T, E>,
106{
107    pending_observables: VecDeque<OE1>,
108    slot: SubscriptionSlot<Subscription<OE1::D>>,
109    is_source_completed: bool,
110}
111
112struct SourceObserver<'or, T, E, OR, OE1, SD>(
113    SubscriptionContext<T, E, OR, Model<'or, T, E, OE1>, SD>,
114)
115where
116    OE1: Observable<'or, T, E>,
117    SD: Disposable;
118
119impl<'or, T, E, OR, OE1, SD> Observer<OE1, E> for SourceObserver<'or, T, E, OR, OE1, SD>
120where
121    T: MaybeSend + 'or,
122    E: MaybeSend + 'or,
123    OR: Observer<T, E> + MaybeSend + 'or,
124    OE1: Observable<'or, T, E> + MaybeSend + 'or,
125    OE1::D: MaybeSend + 'or,
126    SD: Disposable + MaybeSend + 'or,
127{
128    fn on_next(&mut self, value: OE1) -> Flow {
129        let result = self.0.update(|model| {
130            if model.slot.reserve_if_idle() {
131                UpdateOutcome::new(Some(value))
132            } else {
133                model.pending_observables.push_back(value);
134                UpdateOutcome::new(None)
135            }
136        });
137        let observable = match result {
138            Ok(Some(observable)) => observable,
139            Ok(None) => return Flow::Continue,
140            Err(DeliveryStopped) => return Flow::Stop,
141        };
142        let observer = InnerObserver(self.0.clone());
143        let sub = observable.subscribe(observer);
144        // `fill` gives the subscription back when the slot was released while it was being built,
145        // which means the operator already terminated.
146        self.0
147            .update_flow(|model| UpdateOutcome::empty().with_drop_outside(model.slot.fill(sub)))
148    }
149
150    fn on_termination(self, termination: Termination<E>) {
151        match termination {
152            completion @ Termination::Completed => {
153                let _ = self.0.update(|model| {
154                    model.is_source_completed = true;
155                    if model.slot.is_idle() {
156                        // The slot is only possible to be reserved or active when the pending_observables is not empty.
157                        debug_assert!(model.pending_observables.is_empty());
158                        UpdateOutcome::empty().with_termination_event(completion)
159                    } else {
160                        UpdateOutcome::empty().without_events()
161                    }
162                });
163            }
164            error @ Termination::Error(_) => {
165                self.0.send_termination(error);
166            }
167        }
168    }
169}
170
171struct InnerObserver<'or, T, E, OR, OE1, SD>(
172    SubscriptionContext<T, E, OR, Model<'or, T, E, OE1>, SD>,
173)
174where
175    OE1: Observable<'or, T, E>,
176    SD: Disposable;
177
178impl<'or, T, E, OR, OE1, SD> Observer<T, E> for InnerObserver<'or, T, E, OR, OE1, SD>
179where
180    T: MaybeSend + 'or,
181    E: MaybeSend + 'or,
182    OR: Observer<T, E> + MaybeSend + 'or,
183    OE1: Observable<'or, T, E> + MaybeSend + 'or,
184    OE1::D: MaybeSend + 'or,
185    SD: Disposable + MaybeSend + 'or,
186{
187    fn on_next(&mut self, value: T) -> Flow {
188        self.0.send_next(value)
189    }
190
191    fn on_termination(self, termination: Termination<E>) {
192        match termination {
193            Termination::Completed => subscribe_next_observable_until_finished(self.0.clone()),
194            error @ Termination::Error(_) => {
195                self.0.send_termination(error);
196            }
197        }
198    }
199}
200
201fn subscribe_next_observable_until_finished<'or, T, E, OR, OE1, SD>(
202    context: SubscriptionContext<T, E, OR, Model<'or, T, E, OE1>, SD>,
203) where
204    T: MaybeSend + 'or,
205    E: MaybeSend + 'or,
206    OR: Observer<T, E> + MaybeSend + 'or,
207    OE1: Observable<'or, T, E> + MaybeSend + 'or,
208    OE1::D: MaybeSend + 'or,
209    SD: Disposable + MaybeSend + 'or,
210{
211    loop {
212        let result = context.update(|model| {
213            if model.slot.is_reserved() {
214                // Already terminated. Releasing a reserved slot makes the pending fill give its
215                // subscription back; the slot itself holds nothing, so this hands nothing out,
216                // and it is handed out rather than asserted under the lock.
217                let released = model.slot.release();
218                return UpdateOutcome::new(None)
219                    .without_events()
220                    .with_drop_outside(released);
221            }
222            if let Some(observable) = model.pending_observables.pop_front() {
223                UpdateOutcome::new(Some(observable))
224                    .without_events()
225                    .with_drop_outside(model.slot.reserve())
226            } else if model.is_source_completed {
227                UpdateOutcome::new(None)
228                    .with_termination_event(Termination::Completed)
229                    .without_drop_outside()
230            } else {
231                UpdateOutcome::new(None)
232                    .without_events()
233                    .with_drop_outside(model.slot.release())
234            }
235        });
236        let observable = match result {
237            Ok(Some(observable)) => observable,
238            Ok(None) => break,
239            Err(DeliveryStopped) => {
240                break;
241            }
242        };
243        let observer = InnerObserver(context.clone());
244        let sub = observable.subscribe(observer);
245        let result = context.update(|model| {
246            // `fill` gives the subscription back when the slot was released while it was being
247            // built, which means the inner observable already terminated: the loop then goes on
248            // to the next pending observable instead of waiting for this subscription.
249            let unused = model.slot.fill(sub);
250            UpdateOutcome::new(unused.is_none()).with_drop_outside(unused)
251        });
252        match result {
253            Ok(subscribed) => {
254                if subscribed {
255                    break;
256                }
257            }
258            Err(DeliveryStopped) => break,
259        }
260    }
261}
rx_rust/operators/combining/concat_all.rs

rx_rust/operators/combining/
concat_all.rs
