use crate::yielder::Receiver; use futures_core::{FusedStream, Stream}; use pin_project_lite::pin_project; use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; pin_project! { #[doc(hidden)] #[derive(Debug)] pub struct AsyncStream { rx: Receiver, done: bool, #[pin] generator: U, } } impl AsyncStream { #[doc(hidden)] pub fn new(rx: Receiver, generator: U) -> AsyncStream { AsyncStream { rx, done: false, generator, } } } impl FusedStream for AsyncStream where U: Future, { fn is_terminated(&self) -> bool { self.done } } impl Stream for AsyncStream where U: Future, { type Item = T; fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let me = self.project(); if *me.done { return Poll::Ready(None); } let mut dst = None; let res = { let _enter = me.rx.enter(&mut dst); me.generator.poll(cx) }; *me.done = res.is_ready(); if dst.is_some() { return Poll::Ready(dst.take()); } if *me.done { Poll::Ready(None) } else { Poll::Pending } } fn size_hint(&self) -> (usize, Option) { if self.done { (0, Some(0)) } else { (0, None) } } }