rx_rust/operators/others/
observable_future.rs

1use crate::{
2    observable::Observable, operators::others::observable_try_future::ObservableTryFuture,
3    utils::types::MaybeSend,
4};
5use educe::Educe;
6use std::{convert::Infallible, task::Poll};
7
8/// A `Future` of the first item of an Observable that cannot fail.
9///
10/// It resolves with `Some(item)` on the first item, stopping the source right there, and with
11/// `None` when the source completes without one. This is
12/// [`ObservableTryFuture`] without the error it can never carry; see there for how it subscribes
13/// and when it releases the source.
14///
15/// # Examples
16/// ```rust
17/// # #[cfg(not(feature = "tokio-scheduler"))]
18/// # fn main() {}
19/// # #[cfg(feature = "tokio-scheduler")]
20/// #[tokio::main]
21/// async fn main() {
22///     use rx_rust::{
23///         observable::ObservableExt,
24///         operators::creating::{empty::Empty, from_iter::FromIter},
25///     };
26///
27///     let first = FromIter::new([10, 20, 30]).into_future().await;
28///     assert_eq!(first, Some(10));
29///
30///     let none = Empty.with_item_type::<i32>().into_future().await;
31///     assert_eq!(none, None);
32/// }
33/// ```
34#[derive(Educe)]
35#[educe(Debug)]
36pub struct ObservableFuture<'or, T, OE>
37where
38    OE: Observable<'or, T, Infallible>,
39{
40    future: ObservableTryFuture<'or, T, Infallible, OE>,
41}
42
43impl<'or, T, OE> ObservableFuture<'or, T, OE>
44where
45    OE: Observable<'or, T, Infallible>,
46{
47    pub fn new(source: OE) -> Self {
48        Self {
49            future: ObservableTryFuture::new(source),
50        }
51    }
52}
53
54impl<'or, T, OE> Unpin for ObservableFuture<'or, T, OE> where OE: Observable<'or, T, Infallible> {}
55
56impl<'or, T, OE> Future for ObservableFuture<'or, T, OE>
57where
58    T: MaybeSend + 'or,
59    OE: Observable<'or, T, Infallible>,
60{
61    type Output = Option<T>;
62
63    fn poll(
64        mut self: std::pin::Pin<&mut Self>,
65        cx: &mut std::task::Context<'_>,
66    ) -> Poll<Self::Output> {
67        std::pin::Pin::new(&mut self.future).poll(cx).map(|result| {
68            let Ok(value) = result;
69            value
70        })
71    }
72}
rx_rust/operators/others/observable_future.rs

rx_rust/operators/others/
observable_future.rs
