rx_rust/operators/creating/
from_try_future.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    observable::{Observable, Subscription},
4    observer::{Observer, Termination},
5    scheduler::Scheduler,
6};
7use educe::Educe;
8
9/// Converts a `Future` of a `Result` into an Observable.
10/// See <https://reactivex.io/documentation/operators/from.html>
11///
12/// An `Ok` is emitted as the single item, and the Observable then completes; an `Err` terminates it
13/// with that error. This is how a one-shot result, the `Single` of ReactiveX, enters a pipeline.
14/// A future that cannot fail goes through
15/// [`FromFuture`](crate::operators::creating::from_future::FromFuture) instead.
16///
17/// # Examples
18/// ```rust
19/// # #[cfg(not(feature = "tokio-scheduler"))]
20/// # fn main() {}
21/// # #[cfg(feature = "tokio-scheduler")]
22/// #[tokio::main]
23/// async fn main() {
24///     use rx_rust::{
25///         observable::ObservableExt,
26///         observer::Termination,
27///         operators::creating::from_try_future::FromTryFuture,
28///     };
29///     use std::sync::{Arc, Mutex};
30///     use tokio::time::{sleep, Duration};
31///
32///     let values = Arc::new(Mutex::new(Vec::new()));
33///     let terminations = Arc::new(Mutex::new(Vec::new()));
34///     let values_observer = Arc::clone(&values);
35///     let terminations_observer = Arc::clone(&terminations);
36///     let handle = tokio::runtime::Handle::current();
37///
38///     let subscription = FromTryFuture::new(async { Err::<i32, _>("boom") }, handle)
39///         .subscribe_with_callback(
40///             move |value| values_observer.lock().unwrap().push(value),
41///             move |termination| terminations_observer
42///                 .lock()
43///                 .unwrap()
44///                 .push(termination),
45///         );
46///
47///     sleep(Duration::from_millis(10)).await;
48///     drop(subscription);
49///
50///     assert!(values.lock().unwrap().is_empty());
51///     assert_eq!(
52///         &*terminations.lock().unwrap(),
53///         &[Termination::Error("boom")]
54///     );
55/// }
56/// ```
57#[derive(Educe)]
58#[educe(Debug, Clone)]
59pub struct FromTryFuture<FU, S> {
60    future: FU,
61    scheduler: S,
62}
63
64impl<FU, S> FromTryFuture<FU, S> {
65    pub fn new(future: FU, scheduler: S) -> Self {
66        Self { future, scheduler }
67    }
68}
69
70impl<T, E, FU, S> Observable<'static, T, E> for FromTryFuture<FU, S>
71where
72    FU: Future<Output = Result<T, E>> + MaybeSend + 'static,
73    S: Scheduler,
74{
75    type D = S::D;
76
77    fn subscribe(
78        self,
79        mut observer: impl Observer<T, E> + MaybeSend + 'static,
80    ) -> Subscription<Self::D> {
81        self.scheduler.spawn_future(async {
82            match self.future.await {
83                Ok(value) => {
84                    if observer.on_next(value).is_continue() {
85                        observer.on_termination(Termination::Completed);
86                    }
87                }
88                Err(error) => observer.on_termination(Termination::Error(error)),
89            }
90        })
91    }
92}
rx_rust/operators/creating/from_try_future.rs

rx_rust/operators/creating/
from_try_future.rs
