rx_rust/operators/creating/
from_future.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    observable::{Observable, Subscription},
4    observer::{Observer, Termination},
5    scheduler::Scheduler,
6};
7use educe::Educe;
8use std::convert::Infallible;
9
10/// Converts a Future into an Observable.
11/// See <https://reactivex.io/documentation/operators/from.html>
12///
13/// The output is emitted as the single item, and the Observable then completes. A future of a
14/// `Result` goes through [`FromTryFuture`](crate::operators::creating::from_try_future::FromTryFuture)
15/// instead, which turns its `Err` into the error of the Observable.
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_future::FromFuture,
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 = FromFuture::new(async { 7 }, handle).subscribe_with_callback(
39///         move |value| values_observer.lock().unwrap().push(value),
40///         move |termination| terminations_observer
41///             .lock()
42///             .unwrap()
43///             .push(termination),
44///     );
45///
46///     sleep(Duration::from_millis(10)).await;
47///     drop(subscription);
48///
49///     assert_eq!(&*values.lock().unwrap(), &[7]);
50///     assert_eq!(
51///         &*terminations.lock().unwrap(),
52///         &[Termination::Completed]
53///     );
54/// }
55/// ```
56#[derive(Educe)]
57#[educe(Debug, Clone)]
58pub struct FromFuture<FU, S> {
59    future: FU,
60    scheduler: S,
61}
62
63impl<FU, S> FromFuture<FU, S> {
64    pub fn new(future: FU, scheduler: S) -> Self {
65        Self { future, scheduler }
66    }
67}
68
69impl<T, FU, S> Observable<'static, T, Infallible> for FromFuture<FU, S>
70where
71    FU: Future<Output = T> + MaybeSend + 'static,
72    S: Scheduler,
73{
74    type D = S::D;
75
76    fn subscribe(
77        self,
78        mut observer: impl Observer<T, Infallible> + MaybeSend + 'static,
79    ) -> Subscription<Self::D> {
80        self.scheduler.spawn_future(async {
81            let result = self.future.await;
82            if observer.on_next(result).is_continue() {
83                observer.on_termination(Termination::Completed);
84            }
85        })
86    }
87}
rx_rust/operators/creating/from_future.rs

rx_rust/operators/creating/
from_future.rs
