rx_rust/operators/combining/
concat.rs

1use crate::delegate_disposal;
2use crate::disposable::{
3    Disposable, chain_disposal::ChainDisposal, shared_disposal::SharedDisposal,
4};
5use crate::utils::types::MaybeSend;
6use crate::{
7    observable::{Observable, Subscription},
8    observer::{Flow, Observer, Termination},
9};
10use educe::Educe;
11
12/// Concatenates multiple Observables to create an Observable that emits all of the values from the first, then all of the values from the second, and so on.
13/// See <https://reactivex.io/documentation/operators/concat.html>
14///
15/// # Examples
16/// ```rust
17/// use rx_rust::{
18///     observable::ObservableExt,
19///     observer::Termination,
20///     operators::{
21///         combining::concat::Concat,
22///         creating::from_iter::FromIter,
23///     },
24/// };
25///
26/// let mut values = Vec::new();
27/// let mut terminations = Vec::new();
28///
29/// let observable =
30///     Concat::new(FromIter::new(vec![1, 2]), FromIter::new(vec![3, 4]));
31/// observable.subscribe_with_callback(
32///     |value| values.push(value),
33///     |termination| terminations.push(termination),
34/// );
35///
36/// assert_eq!(values, vec![1, 2, 3, 4]);
37/// assert_eq!(terminations, vec![Termination::Completed]);
38/// ```
39#[derive(Educe)]
40#[educe(Debug, Clone)]
41pub struct Concat<OE1, OE2> {
42    source_1: OE1,
43    source_2: OE2,
44}
45
46impl<OE1, OE2> Concat<OE1, OE2> {
47    pub fn new<'or, T, E>(source_1: OE1, source_2: OE2) -> Self
48    where
49        OE1: Observable<'or, T, E>,
50        OE2: Observable<'or, T, E>,
51    {
52        Self { source_1, source_2 }
53    }
54}
55
56delegate_disposal!(
57    Disposal<D1, D2>,
58    ChainDisposal<SharedDisposal<Subscription<D2>>, D1>,
59    where D1: Disposable, D2: Disposable
60);
61
62impl<'or, T, E, OE1, OE2> Observable<'or, T, E> for Concat<OE1, OE2>
63where
64    OE1: Observable<'or, T, E>,
65    OE2: Observable<'or, T, E> + MaybeSend + 'or,
66    OE2::D: MaybeSend + 'or,
67{
68    type D = Disposal<OE1::D, OE2::D>;
69
70    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
71        let sub_2 = SharedDisposal::default();
72        let observer = ConcatObserver {
73            observer,
74            source_2: self.source_2,
75            sub_2: sub_2.clone(),
76        };
77        self.source_1
78            .subscribe(observer)
79            .preceded_by(sub_2)
80            .map_into()
81    }
82}
83
84struct ConcatObserver<OR, OE2, D: Disposable> {
85    observer: OR,
86    source_2: OE2,
87    sub_2: SharedDisposal<Subscription<D>>,
88}
89
90impl<'or, T, E, OR, OE2> Observer<T, E> for ConcatObserver<OR, OE2, OE2::D>
91where
92    OR: Observer<T, E> + MaybeSend + 'or,
93    OE2: Observable<'or, T, E>,
94{
95    fn on_next(&mut self, value: T) -> Flow {
96        self.observer.on_next(value)
97    }
98
99    fn on_termination(self, termination: Termination<E>) {
100        match termination {
101            Termination::Completed => {
102                // `replace` does not run the builder once the subscription was disposed, so the
103                // second source is never subscribed after downstream unsubscribed, and a
104                // subscription built while downstream unsubscribes is disposed right away.
105                self.sub_2
106                    .replace(|| self.source_2.subscribe(self.observer));
107            }
108            error @ Termination::Error(_) => {
109                self.observer.on_termination(error);
110            }
111        }
112    }
113}
rx_rust/operators/combining/concat.rs

rx_rust/operators/combining/
concat.rs
