rx_rust/operators/combining/
merge.rs

1use crate::safe_lock_option_observer;
2use crate::utils::types::{Mutable, NecessarySendSync, Shared};
3use crate::{
4    disposable::subscription::Subscription,
5    observable::Observable,
6    observer::{Observer, Termination},
7    utils::unsub_after_termination::subscribe_unsub_after_termination,
8};
9use educe::Educe;
10use std::sync::atomic::{AtomicBool, Ordering};
11
12/// Combines multiple Observables into a single Observable that emits all of their emissions.
13/// See <https://reactivex.io/documentation/operators/merge.html>
14///
15/// # Examples
16/// ```rust
17/// use rx_rust::{
18///     observable::observable_ext::ObservableExt,
19///     observer::Termination,
20///     operators::{
21///         combining::merge::Merge,
22///         creating::from_iter::FromIter,
23///     },
24/// };
25///
26/// let mut values = Vec::new();
27/// let mut terminations = Vec::new();
28///
29/// let observable = Merge::new(
30///     FromIter::new(vec![1, 3]),
31///     FromIter::new(vec![2, 4]),
32/// );
33/// observable.subscribe_with_callback(
34///     |value| values.push(value),
35///     |termination| terminations.push(termination),
36/// );
37///
38/// assert_eq!(values, vec![1, 3, 2, 4]);
39/// assert_eq!(terminations, vec![Termination::Completed]);
40/// ```
41#[derive(Educe)]
42#[educe(Debug, Clone)]
43pub struct Merge<OE1, OE2> {
44    source_1: OE1,
45    source_2: OE2,
46}
47
48impl<OE1, OE2> Merge<OE1, OE2> {
49    pub fn new<'or, 'sub, T, E>(source_1: OE1, source_2: OE2) -> Self
50    where
51        OE1: Observable<'or, 'sub, T, E>,
52        OE2: Observable<'or, 'sub, T, E>,
53    {
54        Self { source_1, source_2 }
55    }
56}
57
58impl<'or, 'sub, T, E, OE1, OE2> Observable<'or, 'sub, T, E> for Merge<OE1, OE2>
59where
60    OE1: Observable<'or, 'sub, T, E>,
61    OE2: Observable<'or, 'sub, T, E>,
62    'sub: 'or,
63{
64    fn subscribe(self, observer: impl Observer<T, E> + NecessarySendSync + 'or) -> Subscription<'sub> {
65        subscribe_unsub_after_termination(observer, |observer| {
66            let observer = Shared::new(Mutable::new(Some(observer)));
67            let one_is_completed = Shared::new(AtomicBool::new(false));
68            let onserver_1 = MergeObserver {
69                observer: observer.clone(),
70                one_is_completed: one_is_completed.clone(),
71            };
72            let onserver_2 = MergeObserver {
73                observer,
74                one_is_completed,
75            };
76            let subscription_1 = self.source_1.subscribe(onserver_1);
77            let subscription_2 = self.source_2.subscribe(onserver_2);
78            subscription_1 + subscription_2
79        })
80    }
81}
82
83struct MergeObserver<OR> {
84    observer: Shared<Mutable<Option<OR>>>,
85    one_is_completed: Shared<AtomicBool>,
86}
87
88impl<T, E, OR> Observer<T, E> for MergeObserver<OR>
89where
90    OR: Observer<T, E>,
91{
92    fn on_next(&mut self, value: T) {
93        safe_lock_option_observer!(on_next: self.observer, value);
94    }
95
96    fn on_termination(self, termination: Termination<E>) {
97        match termination {
98            Termination::Completed => {
99                if self.one_is_completed.load(Ordering::SeqCst) {
100                    safe_lock_option_observer!(on_termination: self.observer, termination);
101                } else {
102                    self.one_is_completed.store(true, Ordering::SeqCst);
103                }
104            }
105            Termination::Error(_) => {
106                safe_lock_option_observer!(on_termination: self.observer, termination);
107            }
108        }
109    }
110}
rx_rust/operators/combining/merge.rs

rx_rust/operators/combining/
merge.rs
