rx_rust/observer/
boxed_observer.rs

1use super::{Flow, Observer, Termination};
2use crate::utils::types::MaybeSend;
3
4trait ErasedObserver<T, E>: Observer<T, E> {
5    fn on_termination_boxed(self: Box<Self>, termination: Termination<E>);
6}
7
8impl<T, E, OR> ErasedObserver<T, E> for OR
9where
10    OR: Observer<T, E>,
11{
12    fn on_termination_boxed(self: Box<Self>, termination: Termination<E>) {
13        Observer::on_termination(*self, termination);
14    }
15}
16
17cfg_if::cfg_if! {
18    if #[cfg(feature = "single-threaded")] {
19        /// Type-erased observer for single-threaded builds to handle this problem <https://stackoverflow.com/q/46620790/9315497>
20        pub struct BoxedObserver<'or, T, E>(Box<dyn ErasedObserver<T, E> + 'or>);
21    } else {
22        /// Type-erased observer for multi-threaded builds to handle this problem <https://stackoverflow.com/q/46620790/9315497>
23        pub struct BoxedObserver<'or, T, E>(Box<dyn ErasedObserver<T, E> + Send + 'or>);
24    }
25}
26
27impl<'or, T, E> BoxedObserver<'or, T, E> {
28    pub fn new(observer: impl Observer<T, E> + MaybeSend + 'or) -> Self {
29        Self(Box::new(observer))
30    }
31}
32
33impl<T, E> Observer<T, E> for BoxedObserver<'_, T, E> {
34    #[inline]
35    fn on_next(&mut self, value: T) -> Flow {
36        self.0.on_next(value)
37    }
38
39    #[inline]
40    fn on_termination(self, termination: Termination<E>) {
41        self.0.on_termination_boxed(termination);
42    }
43}
44
45impl<T, E> std::fmt::Debug for BoxedObserver<'_, T, E> {
46    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47        f.write_str(std::any::type_name::<Self>())
48    }
49}
rx_rust/observer/boxed_observer.rs

rx_rust/observer/
boxed_observer.rs
