rx_rust/operators/combining/
combine_latest.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;
12
13/// Combines multiple Observables to create an Observable whose values are calculated from the latest values of each of its input Observables.
14/// See <https://reactivex.io/documentation/operators/combinelatest.html>
15///
16/// # Examples
17/// ```rust
18/// use rx_rust::{
19///     observable::ObservableExt,
20///     observer::{Observer, Termination},
21///     operators::combining::combine_latest::CombineLatest,
22///     subject::behavior_subject::BehaviorSubject,
23/// };
24/// use std::convert::Infallible;
25///
26/// let mut values = Vec::new();
27/// let mut terminations = Vec::new();
28///
29/// let mut subject_1 = BehaviorSubject::<'_, i32, Infallible>::new(0);
30/// let mut subject_2 = BehaviorSubject::<'_, i32, Infallible>::new(10);
31///
32/// let subscription =
33///     CombineLatest::new(subject_1.clone(), subject_2.clone()).subscribe_with_callback(
34///         |value| values.push(value),
35///         |termination| terminations.push(termination),
36///     );
37///
38/// subject_1.on_next(1);
39/// subject_2.on_next(11);
40/// subject_1.on_termination(Termination::Completed);
41/// subject_2.on_termination(Termination::Completed);
42/// drop(subscription);
43///
44/// assert_eq!(values, vec![(0, 10), (1, 10), (1, 11)]);
45/// assert_eq!(terminations, vec![Termination::Completed]);
46/// ```
47#[derive(Educe)]
48#[educe(Debug, Clone)]
49pub struct CombineLatest<OE1, OE2> {
50    source_1: OE1,
51    source_2: OE2,
52}
53
54impl<OE1, OE2> CombineLatest<OE1, OE2> {
55    pub fn new<'or, T1, T2, E>(source_1: OE1, source_2: OE2) -> Self
56    where
57        OE1: Observable<'or, T1, E>,
58        OE2: Observable<'or, T2, E>,
59    {
60        Self { source_1, source_2 }
61    }
62}
63
64impl<'or, T1, T2, E, OE1, OE2> Observable<'or, (T1, T2), E> for CombineLatest<OE1, OE2>
65where
66    T1: Clone + MaybeSend + 'or,
67    T2: Clone + MaybeSend + 'or,
68    E: MaybeSend + 'or,
69    OE1: Observable<'or, T1, E>,
70    OE1::D: MaybeSend + 'or,
71    OE2: Observable<'or, T2, E>,
72    OE2::D: MaybeSend + 'or,
73{
74    type D = subscribe_with_context::OwningDisposal<'or>;
75
76    fn subscribe(
77        self,
78        observer: impl Observer<(T1, T2), E> + MaybeSend + 'or,
79    ) -> Subscription<Self::D> {
80        let model = Model {
81            latest_1: None,
82            latest_2: None,
83            should_completed: false,
84        };
85        subscribe_with_context_owning_source(observer, model, |context| {
86            let sub_1 = self.source_1.subscribe(ObserverImpl1(context.clone()));
87            let sub_2 = self.source_2.subscribe(ObserverImpl2(context));
88            sub_1.preceded_by_bound(sub_2)
89        })
90    }
91}
92
93struct Model<T1, T2> {
94    latest_1: Option<T1>,
95    latest_2: Option<T2>,
96    should_completed: bool,
97}
98
99macro_rules! impl_observer {
100    ($name:ident, $t_self:ident, $field_self:ident, $field_other:ident, $combine:expr) => {
101        struct $name<T1, T2, E, OR, D: Disposable>(
102            SubscriptionContext<(T1, T2), E, OR, Model<T1, T2>, D>,
103        );
104
105        impl<T1, T2, E, OR, D> Observer<$t_self, E> for $name<T1, T2, E, OR, D>
106        where
107            T1: Clone,
108            T2: Clone,
109            OR: Observer<(T1, T2), E>,
110            D: Disposable,
111        {
112            fn on_next(&mut self, val: $t_self) -> Flow {
113                self.0.update_flow(|model| {
114                    // The latest value this one replaces is handed back, so that the `Drop` of
115                    // the user's value runs outside the lock. Building the pair still clones
116                    // under it: whether there is a pair to build at all is only known here.
117                    if let Some(other) = &model.$field_other {
118                        let pair = $combine(val.clone(), other.clone());
119                        let replaced = model.$field_self.replace(val);
120                        UpdateOutcome::empty()
121                            .with_drop_outside(replaced)
122                            .with_next_event(pair)
123                    } else {
124                        let replaced = model.$field_self.replace(val);
125                        UpdateOutcome::empty()
126                            .with_drop_outside(replaced)
127                            .without_events()
128                    }
129                })
130            }
131
132            fn on_termination(self, termination: Termination<E>) {
133                let _ = self.0.update(|model| match termination {
134                    completion @ Termination::Completed => {
135                        if model.should_completed || model.$field_self.is_none() {
136                            UpdateOutcome::empty().with_termination_event(completion)
137                        } else {
138                            model.should_completed = true;
139                            UpdateOutcome::empty().without_events()
140                        }
141                    }
142                    error @ Termination::Error(_) => {
143                        UpdateOutcome::empty().with_termination_event(error)
144                    }
145                });
146            }
147        }
148    };
149}
150
151impl_observer!(ObserverImpl1, T1, latest_1, latest_2, |v, o| (v, o));
152impl_observer!(ObserverImpl2, T2, latest_2, latest_1, |v, o| (o, v));
rx_rust/operators/combining/combine_latest.rs

rx_rust/operators/combining/
combine_latest.rs
