rx_rust/operators/utility/
dematerialize.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    observable::Observable,
4    observable::Subscription,
5    observer::{Event, Flow, Observer, Termination},
6    utils::subscribe_with_auto_dispose_on_termination::subscribe_with_auto_dispose_on_termination,
7};
8use educe::Educe;
9use std::convert::Infallible;
10
11/// Converts an Observable that emits `Event` objects into a "live" Observable that emits the items and notifications embedded in those `Event` objects.
12/// See <https://reactivex.io/documentation/operators/materialize-dematerialize.html>
13///
14/// # Examples
15/// ```rust
16/// use rx_rust::{
17///     observable::ObservableExt,
18///     observer::Termination,
19///     operators::{
20///         creating::from_iter::FromIter,
21///         utility::{
22///             dematerialize::Dematerialize,
23///             materialize::Materialize,
24///         },
25///     },
26/// };
27///
28/// let mut values = Vec::new();
29/// let mut terminations = Vec::new();
30///
31/// Dematerialize::new(Materialize::new(FromIter::new(vec![1, 2])))
32///     .subscribe_with_callback(
33///         |value| values.push(value),
34///         |termination| terminations.push(termination),
35///     );
36///
37/// assert_eq!(values, vec![1, 2]);
38/// assert_eq!(terminations, vec![Termination::Completed]);
39/// ```
40#[derive(Educe)]
41#[educe(Debug, Clone)]
42pub struct Dematerialize<OE>(OE);
43
44impl<OE> Dematerialize<OE> {
45    pub fn new(source: OE) -> Self {
46        Self(source)
47    }
48}
49
50impl<'or, T, E, OE> Observable<'or, T, E> for Dematerialize<OE>
51where
52    OE: Observable<'or, Event<T, E>, Infallible>,
53    OE::D: MaybeSend + 'or,
54{
55    type D = crate::utils::subscribe_with_auto_dispose_on_termination::Disposal<OE::D>;
56
57    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
58        subscribe_with_auto_dispose_on_termination(observer, |observer| {
59            self.0.subscribe(DematerializeObserver(Some(observer)))
60        })
61    }
62}
63
64struct DematerializeObserver<OR>(Option<OR>);
65
66impl<T, E, OR> Observer<Event<T, E>, Infallible> for DematerializeObserver<OR>
67where
68    OR: Observer<T, E>,
69{
70    fn on_next(&mut self, value: Event<T, E>) -> Flow {
71        match value {
72            Event::Next(value) => match self.0.as_mut() {
73                Some(observer) => {
74                    let flow = observer.on_next(value);
75                    if flow.is_stop() {
76                        drop(self.0.take());
77                    }
78                    flow
79                }
80                // The materialized termination already ended the stream downstream.
81                None => Flow::Stop,
82            },
83            Event::Termination(termination) => {
84                if let Some(observer) = self.0.take() {
85                    observer.on_termination(termination);
86                }
87                // The stream ends with the value that carried the termination, so whatever the
88                // source has left is of no use anymore.
89                Flow::Stop
90            }
91        }
92    }
93
94    fn on_termination(mut self, termination: Termination<Infallible>) {
95        match termination {
96            Termination::Completed => {
97                if let Some(observer) = self.0.take() {
98                    observer.on_termination(Termination::Completed);
99                }
100            }
101            // `Infallible` is uninhabited, so the compiler proves this arm unreachable.
102            Termination::Error(error) => match error {},
103        }
104    }
105}
rx_rust/operators/utility/dematerialize.rs

rx_rust/operators/utility/
dematerialize.rs
