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

rx_rust/operators/creating/
from_try_stream.rs
