f

Author: jean-airoldie

futures-util/src/stream/stream/buffered.rs

@@ -18,6 +18,7 @@ where
     stream: Fuse<St>,
     in_progress_queue: FuturesOrdered<St::Item>,
     max: usize,
+    is_terminated: bool,
 }
 
 impl<St> Unpin for Buffered<St>
@@ -36,6 +37,7 @@ where
             .field("stream", &self.stream)
             .field("in_progress_queue", &self.in_progress_queue)
             .field("max", &self.max)
+            .field("is_terminated", &self.is_terminated)
             .finish()
     }
 }
@@ -46,7 +48,7 @@ where
     St::Item: Future,
 {
     fn is_terminated(&self) -> bool {
-        self.stream.is_terminated() && self.in_progress_queue.is_empty()
+        self.is_terminated
     }
 }
 
@@ -57,12 +59,14 @@ where
 {
     unsafe_pinned!(stream: Fuse<St>);
     unsafe_unpinned!(in_progress_queue: FuturesOrdered<St::Item>);
+    unsafe_unpinned!(is_terminated: bool);
 
     pub(super) fn new(stream: St, n: usize) -> Buffered<St> {
         Buffered {
             stream: super::Fuse::new(stream),
             in_progress_queue: FuturesOrdered::new(),
             max: n,
+            is_terminated: false,
         }
     }
 
@@ -127,6 +131,9 @@ where
 
         // If more values are still coming from the stream, we're not done yet
         if self.stream.is_done() {
+            // We yield a `None`, so now the stream is considered terminated.
+            *self.as_mut().is_terminated() = true;
+
             Poll::Ready(None)
         } else {
             Poll::Pending