rx_rust/operators/combining/
zip.rs

1use crate::utils::serialized_delivery::UpdateOutcome;
2use crate::utils::subscribe_with_context::{
3    self, SubscriptionContext, subscribe_with_context_owning_source,
4};
5use crate::utils::types::MaybeSend;
6use crate::{
7    disposable::Disposable,
8    observable::{Observable, Subscription},
9    observer::{Flow, Observer, Termination},
10};
11use educe::Educe;
12use std::collections::VecDeque;
13
14/// Combines the emissions of multiple Observables together via a specified function and emits single items for each combination based on the sequence of their emissions.
15/// See <https://reactivex.io/documentation/operators/zip.html>
16///
17/// # Examples
18/// ```rust
19/// use rx_rust::{
20///     observable::ObservableExt,
21///     observer::Termination,
22///     operators::{
23///         combining::zip::Zip,
24///         creating::from_iter::FromIter,
25///     },
26/// };
27///
28/// let mut values = Vec::new();
29/// let mut terminations = Vec::new();
30///
31/// let observable = Zip::new(
32///     FromIter::new(vec![1, 2]),
33///     FromIter::new(vec![10, 20]),
34/// );
35/// observable.subscribe_with_callback(
36///     |value| values.push(value),
37///     |termination| terminations.push(termination),
38/// );
39///
40/// assert_eq!(values, vec![(1, 10), (2, 20)]);
41/// assert_eq!(terminations, vec![Termination::Completed]);
42/// ```
43#[derive(Educe)]
44#[educe(Debug, Clone)]
45pub struct Zip<OE1, OE2> {
46    source_1: OE1,
47    source_2: OE2,
48}
49
50impl<OE1, OE2> Zip<OE1, OE2> {
51    pub fn new<'or, T1, T2, E>(source_1: OE1, source_2: OE2) -> Self
52    where
53        OE1: Observable<'or, T1, E>,
54        OE2: Observable<'or, T2, E>,
55    {
56        Self { source_1, source_2 }
57    }
58}
59
60impl<'or, T1, T2, E, OE1, OE2> Observable<'or, (T1, T2), E> for Zip<OE1, OE2>
61where
62    T1: MaybeSend + 'or,
63    T2: MaybeSend + 'or,
64    E: MaybeSend + 'or,
65    OE1: Observable<'or, T1, E>,
66    OE1::D: MaybeSend + 'or,
67    OE2: Observable<'or, T2, E>,
68    OE2::D: MaybeSend + 'or,
69{
70    type D = subscribe_with_context::OwningDisposal<'or>;
71
72    fn subscribe(
73        self,
74        observer: impl Observer<(T1, T2), E> + MaybeSend + 'or,
75    ) -> Subscription<Self::D> {
76        let model = Model {
77            first: (VecDeque::new(), false),
78            second: (VecDeque::new(), false),
79        };
80        subscribe_with_context_owning_source(observer, model, |context| {
81            let subscription_1 = self.source_1.subscribe(ZipObserver1(context.clone()));
82            let subscription_2 = self.source_2.subscribe(ZipObserver2(context));
83            subscription_1.preceded_by_bound(subscription_2)
84        })
85    }
86}
87
88struct Model<T1, T2> {
89    first: (VecDeque<T1>, bool),  // bool means completed
90    second: (VecDeque<T2>, bool), // bool means completed
91}
92
93macro_rules! impl_zip_observer {
94    ($name:ident, $input_t:ty, $this_field:ident, $other_field:ident, $make_pair:expr) => {
95        struct $name<T1, T2, E, OR, D: Disposable>(
96            SubscriptionContext<(T1, T2), E, OR, Model<T1, T2>, D>,
97        );
98
99        impl<T1, T2, E, OR, D> Observer<$input_t, E> for $name<T1, T2, E, OR, D>
100        where
101            OR: Observer<(T1, T2), E>,
102            D: Disposable,
103        {
104            fn on_next(&mut self, value: $input_t) -> Flow {
105                self.0.update_flow(|model| {
106                    if let Some(other) = model.$other_field.0.pop_front() {
107                        if model.$other_field.1 && model.$other_field.0.is_empty() {
108                            UpdateOutcome::empty().with_next_and_termination_events(
109                                $make_pair(value, other),
110                                Termination::Completed,
111                            )
112                        } else {
113                            UpdateOutcome::empty().with_next_event($make_pair(value, other))
114                        }
115                    } else {
116                        model.$this_field.0.push_back(value);
117                        UpdateOutcome::empty().without_events()
118                    }
119                })
120            }
121
122            fn on_termination(self, termination: Termination<E>) {
123                match termination {
124                    completion @ Termination::Completed => {
125                        let _ = self.0.update(|model| {
126                            model.$this_field.1 = true;
127                            if model.$this_field.0.is_empty() {
128                                UpdateOutcome::empty().with_termination_event(completion)
129                            } else {
130                                UpdateOutcome::empty().without_events()
131                            }
132                        });
133                    }
134                    error @ Termination::Error(_) => {
135                        self.0.send_termination(error);
136                    }
137                };
138            }
139        }
140    };
141}
142
143impl_zip_observer!(ZipObserver1, T1, first, second, |this, other| (this, other));
144impl_zip_observer!(ZipObserver2, T2, second, first, |this, other| (other, this));
rx_rust/operators/combining/zip.rs

rx_rust/operators/combining/
zip.rs
