use core::fmt; use core::pin::Pin; use futures_core::future::TryFuture; use futures_core::stream::Stream; use futures_core::task::{Context, Poll}; use pin_utils::{unsafe_pinned, unsafe_unpinned}; /// Creates a `TryStream` from a seed and a closure returning a `TryFuture`. /// /// This function is the dual for the `TryStream::try_fold()` adapter: while /// `TryStream::try_fold()` reduces a `TryStream` to one single value, /// `try_unfold()` creates a `TryStream` from a seed value. /// /// `try_unfold()` will call the provided closure with the provided seed, then /// wait for the returned `TryFuture` to complete with `(a, b)`. It will then /// yield the value `a`, and use `b` as the next internal state. /// /// If the closure returns `None` instead of `Some(TryFuture)`, then the /// `try_unfold()` will stop producing items and return `Poll::Ready(None)` in /// future calls to `poll()`. /// /// In case of error generated by the returned `TryFuture`, the error will be /// returned by the `TryStream`. The `TryStream` will then yield /// `Poll::Ready(None)` in future calls to `poll()`. /// /// This function can typically be used when wanting to go from the "world of /// futures" to the "world of streams": the provided closure can build a /// `TryFuture` using other library functions working on futures, and /// `try_unfold()` will turn it into a `TryStream` by repeating the operation. /// /// # Example /// /// ``` /// # #[derive(Debug, PartialEq)] /// # struct SomeError; /// # futures::executor::block_on(async { /// use futures::stream::{self, TryStreamExt}; /// /// let stream = stream::try_unfold(0, |state| async move { /// if state < 0 { /// return Err(SomeError); /// } /// /// if state <= 2 { /// let next_state = state + 1; /// let yielded = state * 2; /// Ok(Some((yielded, next_state))) /// } else { /// Ok(None) /// } /// }); /// /// let result: Result, _> = stream.try_collect().await; /// assert_eq!(result, Ok(vec![0, 2, 4])); /// # }); /// ``` pub fn try_unfold(init: T, f: F) -> TryUnfold where F: FnMut(T) -> Fut, Fut: TryFuture>, { TryUnfold { f, state: Some(init), fut: None, } } /// Stream for the [`try_unfold`] function. #[must_use = "streams do nothing unless polled"] pub struct TryUnfold { f: F, state: Option, fut: Option, } impl Unpin for TryUnfold {} impl fmt::Debug for TryUnfold where T: fmt::Debug, Fut: fmt::Debug, { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("TryUnfold") .field("state", &self.state) .field("fut", &self.fut) .finish() } } impl TryUnfold { unsafe_unpinned!(f: F); unsafe_unpinned!(state: Option); unsafe_pinned!(fut: Option); } impl Stream for TryUnfold where F: FnMut(T) -> Fut, Fut: TryFuture>, { type Item = Result; fn poll_next( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll>> { if let Some(state) = self.as_mut().state().take() { let fut = (self.as_mut().f())(state); self.as_mut().fut().set(Some(fut)); } match self.as_mut().fut().as_pin_mut() { None => { // The future previously errored Poll::Ready(None) } Some(fut) => { let step = ready!(fut.try_poll(cx)); self.as_mut().fut().set(None); match step { Ok(Some((item, next_state))) => { *self.as_mut().state() = Some(next_state); Poll::Ready(Some(Ok(item))) } Ok(None) => Poll::Ready(None), Err(e) => Poll::Ready(Some(Err(e))), } } } } }