use core::pin::Pin; use futures_core::stream::{FusedStream, Stream, TryStream}; use futures_core::task::{Context, Poll}; #[cfg(feature = "sink")] use futures_sink::Sink; use pin_project::{pin_project, project}; /// Stream for the [`try_flatten`](super::TryStreamExt::try_flatten) method. #[pin_project] #[derive(Debug)] #[must_use = "streams do nothing unless polled"] pub struct TryFlatten where St: TryStream, { #[pin] stream: St, #[pin] next: Option, } impl TryFlatten where St: TryStream, St::Ok: TryStream, ::Error: From, { pub(super) fn new(stream: St) -> Self { Self { stream, next: None } } delegate_access_inner!(stream, St, ()); } impl FusedStream for TryFlatten where St: TryStream + FusedStream, St::Ok: TryStream, ::Error: From, { fn is_terminated(&self) -> bool { self.next.is_none() && self.stream.is_terminated() } } impl Stream for TryFlatten where St: TryStream, St::Ok: TryStream, ::Error: From, { type Item = Result<::Ok, ::Error>; #[project] fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { #[project] let TryFlatten { mut stream, mut next } = self.project(); Poll::Ready(loop { if let Some(s) = next.as_mut().as_pin_mut() { if let Some(item) = ready!(s.try_poll_next(cx)?) { break Some(Ok(item)); } else { next.set(None); } } else if let Some(s) = ready!(stream.as_mut().try_poll_next(cx)?) { next.set(Some(s)); } else { break None; } }) } } // Forwarding impl of Sink from the underlying stream #[cfg(feature = "sink")] impl Sink for TryFlatten where S: TryStream + Sink, { type Error = >::Error; delegate_sink!(stream, Item); }