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(
65        self,
66        observer: impl Observer<T, E> + NecessarySendSync + 'or,
67    ) -> Subscription<'sub> {
68        subscribe_unsub_after_termination(observer, |observer| {
69            let observer = Shared::new(Mutable::new(Some(observer)));
70            let one_is_completed = Shared::new(AtomicBool::new(false));
71            let onserver_1 = MergeObserver {
72                observer: observer.clone(),
73                one_is_completed: one_is_completed.clone(),
74            };
75            let onserver_2 = MergeObserver {
76                observer,
77                one_is_completed,
78            };
79            let subscription_1 = self.source_1.subscribe(onserver_1);
80            let subscription_2 = self.source_2.subscribe(onserver_2);
81            subscription_1 + subscription_2
82        })
83    }
84}
85
86struct MergeObserver<OR> {
87    observer: Shared<Mutable<Option<OR>>>,
88    one_is_completed: Shared<AtomicBool>,
89}
90
91impl<T, E, OR> Observer<T, E> for MergeObserver<OR>
92where
93    OR: Observer<T, E>,
94{
95    fn on_next(&mut self, value: T) {
96        safe_lock_option_observer!(on_next: self.observer, value);
97    }
98
99    fn on_termination(self, termination: Termination<E>) {
100        match termination {
101            Termination::Completed => {
102                if self.one_is_completed.load(Ordering::SeqCst) {
103                    safe_lock_option_observer!(on_termination: self.observer, termination);
104                } else {
105                    self.one_is_completed.store(true, Ordering::SeqCst);
106                }
107            }
108            Termination::Error(_) => {
109                safe_lock_option_observer!(on_termination: self.observer, termination);
110            }
111        }
112    }
113}
rx_rust/operators/combining/merge.rs

rx_rust/operators/combining/
merge.rs
