rx_rust/operators/creating/
from_stream.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    observable::{Observable, Subscription},
4    observer::{Flow, Observer, Termination},
5    scheduler::Scheduler,
6};
7use educe::Educe;
8use futures::Stream;
9use std::convert::Infallible;
10
11/// Converts a `Stream` into an Observable.
12/// See <https://reactivex.io/documentation/operators/from.html>
13///
14/// # Examples
15/// ```rust
16/// # #[cfg(not(feature = "tokio-scheduler"))]
17/// # fn main() {}
18/// # #[cfg(feature = "tokio-scheduler")]
19/// #[tokio::main]
20/// async fn main() {
21///     use futures::stream;
22///     use rx_rust::{
23///         observable::ObservableExt,
24///         observer::Termination,
25///         operators::creating::from_stream::FromStream,
26///     };
27///     use std::sync::{Arc, Mutex};
28///     use tokio::time::{sleep, Duration};
29///
30///     let handle = tokio::runtime::Handle::current();
31///     let values = Arc::new(Mutex::new(Vec::new()));
32///     let terminations = Arc::new(Mutex::new(Vec::new()));
33///     let values_observer = Arc::clone(&values);
34///     let terminations_observer = Arc::clone(&terminations);
35///     let stream = stream::iter([10, 20]);
36///
37///     let subscription = FromStream::new(stream, handle).subscribe_with_callback(
38///         move |value| values_observer.lock().unwrap().push(value),
39///         move |termination| terminations_observer
40///             .lock()
41///             .unwrap()
42///             .push(termination),
43///     );
44///
45///     sleep(Duration::from_millis(10)).await;
46///     drop(subscription);
47///
48///     assert_eq!(&*values.lock().unwrap(), &[10, 20]);
49///     assert_eq!(
50///         &*terminations.lock().unwrap(),
51///         &[Termination::Completed]
52///     );
53/// }
54/// ```
55#[derive(Educe)]
56#[educe(Debug, Clone)]
57pub struct FromStream<SM, S> {
58    stream: SM,
59    scheduler: S,
60}
61
62impl<SM, S> FromStream<SM, S> {
63    pub fn new(stream: SM, scheduler: S) -> Self {
64        Self { stream, scheduler }
65    }
66}
67
68impl<T, SM, S> Observable<'static, T, Infallible> for FromStream<SM, S>
69where
70    SM: Stream<Item = T> + MaybeSend + 'static,
71    S: Scheduler,
72{
73    type D = S::D;
74
75    fn subscribe(
76        self,
77        observer: impl Observer<T, Infallible> + MaybeSend + 'static,
78    ) -> Subscription<Self::D> {
79        let mut observer = Some(observer);
80        self.scheduler
81            .schedule_stream(self.stream, move |result| match result {
82                Some(value) => {
83                    let flow = match observer.as_mut() {
84                        Some(observer) => observer.on_next(value),
85                        None => Flow::Stop,
86                    };
87                    if flow.is_stop() {
88                        // The observer ended its own stream: release it here and tell the
89                        // scheduler to stop polling the stream, so an infinite one is not driven
90                        // for values that have nothing to be delivered to.
91                        drop(observer.take());
92                    }
93                    flow.is_continue()
94                }
95                None => {
96                    if let Some(observer) = observer.take() {
97                        observer.on_termination(Termination::Completed)
98                    }
99                    false
100                }
101            })
102    }
103}
rx_rust/operators/creating/from_stream.rs

rx_rust/operators/creating/
from_stream.rs
