rx_rust/observable/
cloneable_boxed_observable.rs

1use super::{Observable, Observer};
2use crate::{
3    disposable::{Disposable, DisposableExt, boxed_disposal::BoxedDisposal},
4    observable::Subscription,
5    observer::boxed_observer::BoxedObserver,
6    utils::types::{MaybeSend, MaybeSync, Shared},
7};
8use educe::Educe;
9
10trait ErasedCloneableObservable<'or, 'sub, 'oe, T, E> {
11    fn subscribe_cloned(
12        &self,
13        observer: BoxedObserver<'or, T, E>,
14    ) -> Subscription<BoxedDisposal<'sub>>;
15}
16
17impl<'or, 'sub, 'oe, T, E, OE> ErasedCloneableObservable<'or, 'sub, 'oe, T, E> for OE
18where
19    T: 'or,
20    E: 'or,
21    OE: Observable<'or, T, E> + Clone + MaybeSend + 'oe,
22    OE::D: Disposable + MaybeSend + 'sub,
23{
24    fn subscribe_cloned(
25        &self,
26        observer: BoxedObserver<'or, T, E>,
27    ) -> Subscription<BoxedDisposal<'sub>> {
28        Subscription::new(self.clone().subscribe(observer).into_boxed())
29    }
30}
31
32cfg_if::cfg_if! {
33    if #[cfg(feature = "single-threaded")] {
34        /// Cloneable `BoxedObservable` for single-threaded builds.
35        #[derive(Educe)]
36        #[educe(Clone)]
37        pub struct CloneableBoxedObservable<'or, 'sub, 'oe, T, E>(
38            Shared<dyn ErasedCloneableObservable<'or, 'sub, 'oe, T, E> + 'oe>,
39        );
40    } else {
41        /// Cloneable `BoxedObservable` for multi-threaded builds.
42        #[derive(Educe)]
43        #[educe(Clone)]
44        pub struct CloneableBoxedObservable<'or, 'sub, 'oe, T, E>(
45            Shared<dyn ErasedCloneableObservable<'or, 'sub, 'oe, T, E> + Send + Sync + 'oe>,
46        );
47    }
48}
49
50impl<'or, 'sub, 'oe, T, E> CloneableBoxedObservable<'or, 'sub, 'oe, T, E> {
51    pub fn new(
52        observable: impl Observable<'or, T, E, D = impl Disposable + MaybeSend + 'sub>
53        + Clone
54        + MaybeSend
55        + MaybeSync
56        + 'oe,
57    ) -> Self
58    where
59        T: 'or,
60        E: 'or,
61    {
62        Self(Shared::new(observable))
63    }
64}
65
66impl<'or, 'sub, T, E> Observable<'or, T, E> for CloneableBoxedObservable<'or, 'sub, '_, T, E> {
67    type D = BoxedDisposal<'sub>;
68
69    #[inline]
70    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
71        self.0.subscribe_cloned(BoxedObserver::new(observer))
72    }
73}
74
75impl<T, E> std::fmt::Debug for CloneableBoxedObservable<'_, '_, '_, T, E> {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        f.write_str(std::any::type_name::<Self>())
78    }
79}
rx_rust/observable/cloneable_boxed_observable.rs

rx_rust/observable/
cloneable_boxed_observable.rs
