use crate::yielder::Receiver; use futures_core::{FusedStream, Stream}; use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; #[doc(hidden)] #[derive(Debug)] pub struct AsyncStream { rx: Receiver, done: bool, 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> { unsafe { let me = Pin::get_unchecked_mut(self); if me.done { return Poll::Ready(None); } let mut dst = None; let res = { let _enter = me.rx.enter(&mut dst); Pin::new_unchecked(&mut 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) } } }