rx_rust/operators/others/
observable_try_future.rs

1use crate::{
2    observable::{Observable, Subscription},
3    observer::{Flow, Observer, Termination},
4    utils::mutable::{Mutable, MutableHelper},
5    utils::types::{MaybeSend, Shared},
6};
7use educe::Educe;
8use std::task::{Poll, Waker};
9
10#[derive(Educe)]
11#[educe(Debug)]
12struct ObservableTryFutureContext<T, E> {
13    result: Option<Result<Option<T>, E>>,
14    waker: Option<Waker>,
15}
16
17/// A `Future` of the first item of an Observable.
18///
19/// It resolves with `Ok(Some(item))` on the first item, stopping the source right there, with
20/// `Ok(None)` when the source completes without one, and with `Err(error)` when the source
21/// fails first. That makes its output the `Maybe` of ReactiveX; an operator that always emits,
22/// such as `collect` or `reduce`, in front of it gives a `Single`, and `last` picks the last item
23/// instead of the first. A source that cannot fail goes through
24/// [`ObservableFuture`](crate::operators::others::observable_future::ObservableFuture) instead.
25///
26/// Like any future, it does nothing until it is polled: that first poll is what subscribes to the
27/// source. The subscription is dropped as soon as the future resolves, and dropping the future
28/// before that disposes it.
29///
30/// # Examples
31/// ```rust
32/// # #[cfg(not(feature = "tokio-scheduler"))]
33/// # fn main() {}
34/// # #[cfg(feature = "tokio-scheduler")]
35/// #[tokio::main]
36/// async fn main() {
37///     use rx_rust::{
38///         observable::ObservableExt,
39///         operators::creating::{from_iter::FromIter, throw::Throw},
40///     };
41///
42///     let first = FromIter::new([10, 20, 30])
43///         .with_error_type::<&str>()
44///         .into_try_future()
45///         .await;
46///     assert_eq!(first, Ok(Some(10)));
47///
48///     let failed = Throw::new("boom").with_item_type::<i32>().into_try_future().await;
49///     assert_eq!(failed, Err("boom"));
50/// }
51/// ```
52#[derive(Educe)]
53#[educe(Debug)]
54pub struct ObservableTryFuture<'or, T, E, OE>
55where
56    OE: Observable<'or, T, E>,
57{
58    source: Option<OE>,
59    sub: Option<Subscription<OE::D>>,
60    context: Shared<Mutable<ObservableTryFutureContext<T, E>>>,
61}
62
63impl<'or, T, E, OE> ObservableTryFuture<'or, T, E, OE>
64where
65    OE: Observable<'or, T, E>,
66{
67    pub fn new(source: OE) -> Self {
68        Self {
69            source: Some(source),
70            sub: None,
71            context: Shared::new(Mutable::new(ObservableTryFutureContext {
72                result: None,
73                waker: None,
74            })),
75        }
76    }
77}
78
79impl<'or, T, E, OE> Unpin for ObservableTryFuture<'or, T, E, OE> where OE: Observable<'or, T, E> {}
80
81impl<'or, T, E, OE> Future for ObservableTryFuture<'or, T, E, OE>
82where
83    T: MaybeSend + 'or,
84    E: MaybeSend + 'or,
85    OE: Observable<'or, T, E>,
86{
87    type Output = Result<Option<T>, E>;
88
89    fn poll(
90        mut self: std::pin::Pin<&mut Self>,
91        cx: &mut std::task::Context<'_>,
92    ) -> Poll<Self::Output> {
93        if let Some(source) = self.source.take() {
94            let observer = ObservableTryFutureObserver {
95                context: self.context.clone(),
96            };
97            let sub = source.subscribe(observer);
98            self.sub = Some(sub);
99        }
100
101        // The waker this one replaces is handed back, because dropping a `Waker` runs the
102        // external code of its vtable, which must not run under the lock. Once the result is in,
103        // no waker is kept: nothing is left to wake.
104        let (result, previous_waker) =
105            self.context
106                .with_mut(|context| match context.result.take() {
107                    Some(result) => (Some(result), context.waker.take()),
108                    None => (None, context.waker.replace(cx.waker().clone())),
109                });
110        drop(previous_waker); // Drop outside the lock to avoid potential deadlock
111        match result {
112            Some(result) => {
113                // The future is over, so the source is released now instead of whenever the
114                // future itself is dropped.
115                self.sub = None;
116                Poll::Ready(result)
117            }
118            None => Poll::Pending,
119        }
120    }
121}
122
123struct ObservableTryFutureObserver<T, E> {
124    context: Shared<Mutable<ObservableTryFutureContext<T, E>>>,
125}
126
127impl<T, E> ObservableTryFutureObserver<T, E> {
128    fn resolve(&self, result: Result<Option<T>, E>) {
129        // The waker is taken under the lock and woken after it is released, because waking runs
130        // external code, which must not run under the lock. The result this one replaces is
131        // dropped outside it too; there is none unless the source breaks its contract, since a
132        // stopped or terminated observer receives nothing more.
133        let (waker, replaced) = self
134            .context
135            .with_mut(|context| (context.waker.take(), context.result.replace(result)));
136        drop(replaced);
137        if let Some(waker) = waker {
138            waker.wake();
139        }
140    }
141}
142
143impl<T, E> Observer<T, E> for ObservableTryFutureObserver<T, E> {
144    fn on_next(&mut self, value: T) -> Flow {
145        self.resolve(Ok(Some(value)));
146        // The first item is all the future wants: the source stops here and does not terminate
147        // this observer, which has already resolved the future.
148        Flow::Stop
149    }
150
151    fn on_termination(self, termination: Termination<E>) {
152        self.resolve(match termination {
153            Termination::Completed => Ok(None),
154            Termination::Error(error) => Err(error),
155        });
156    }
157}
rx_rust/operators/others/observable_try_future.rs

rx_rust/operators/others/
observable_try_future.rs
