rx_rust/observable/
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,
7};
8
9trait ErasedObservable<'or, 'sub, T, E> {
10    fn subscribe_boxed(
11        self: Box<Self>,
12        observer: BoxedObserver<'or, T, E>,
13    ) -> Subscription<BoxedDisposal<'sub>>;
14}
15
16impl<'or, 'sub, T, E, OE> ErasedObservable<'or, 'sub, T, E> for OE
17where
18    T: 'or,
19    E: 'or,
20    OE: Observable<'or, T, E>,
21    OE::D: Disposable + MaybeSend + 'sub,
22{
23    fn subscribe_boxed(
24        self: Box<Self>,
25        observer: BoxedObserver<'or, T, E>,
26    ) -> Subscription<BoxedDisposal<'sub>> {
27        Subscription::new(Observable::subscribe(*self, observer).into_boxed())
28    }
29}
30
31cfg_if::cfg_if! {
32    if #[cfg(feature = "single-threaded")] {
33        /// Type-erased observable for single-threaded builds to handle this problem <https://stackoverflow.com/q/46620790/9315497>
34        pub struct BoxedObservable<'or, 'sub, 'oe, T, E>(
35            Box<dyn ErasedObservable<'or, 'sub, T, E> + 'oe>,
36        );
37    } else {
38        /// Type-erased observable for multi-threaded builds to handle this problem <https://stackoverflow.com/q/46620790/9315497>
39        pub struct BoxedObservable<'or, 'sub, 'oe, T, E>(
40            Box<dyn ErasedObservable<'or, 'sub, T, E> + Send + 'oe>,
41        );
42    }
43}
44
45impl<'or, 'sub, 'oe, T, E> BoxedObservable<'or, 'sub, 'oe, T, E> {
46    pub fn new(
47        observable: impl Observable<'or, T, E, D = impl Disposable + MaybeSend + 'sub> + MaybeSend + 'oe,
48    ) -> Self
49    where
50        T: 'or,
51        E: 'or,
52    {
53        Self(Box::new(observable))
54    }
55}
56
57impl<'or, 'sub, T, E> Observable<'or, T, E> for BoxedObservable<'or, 'sub, '_, T, E> {
58    type D = BoxedDisposal<'sub>;
59
60    #[inline]
61    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
62        self.0.subscribe_boxed(BoxedObserver::new(observer))
63    }
64}
65
66impl<T, E> std::fmt::Debug for BoxedObservable<'_, '_, '_, T, E> {
67    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
68        f.write_str(std::any::type_name::<Self>())
69    }
70}
rx_rust/observable/boxed_observable.rs

rx_rust/observable/
boxed_observable.rs
