Add FusedStream implementations

futures-util/src/sink/buffer.rs

@@ -1,4 +1,4 @@
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
@@ -78,6 +78,12 @@ impl<S, Item> Stream for Buffer<S, Item> where S: Sink<Item> + Stream {
     }
 }
 
+impl<S, Item> FusedStream for Buffer<S, Item> where S: Sink<Item> + FusedStream {
+    fn is_terminated(&self) -> bool {
+        self.sink.is_terminated()
+    }
+}
+
 impl<Si: Sink<Item>, Item> Sink<Item> for Buffer<Si, Item> {
     type Error = Si::Error;
 

futures-util/src/sink/err_into.rs

@@ -1,6 +1,6 @@
 use crate::sink::{SinkExt, SinkMapErr};
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use futures_sink::{Sink};
 use pin_utils::unsafe_pinned;
@@ -57,6 +57,7 @@ impl<Si, Item, E> Sink<Item> for SinkErrInto<Si, Item, E>
     delegate_sink!(sink, Item);
 }
 
+// Forwarding impl of Stream from the underlying sink
 impl<S, Item, E> Stream for SinkErrInto<S, Item, E>
     where S: Sink<Item> + Stream,
           S::Error: Into<E>
@@ -70,3 +71,12 @@ impl<S, Item, E> Stream for SinkErrInto<S, Item, E>
         self.sink().poll_next(cx)
     }
 }
+
+impl<S, Item, E> FusedStream for SinkErrInto<S, Item, E>
+    where S: Sink<Item> + FusedStream,
+          S::Error: Into<E>
+{
+    fn is_terminated(&self) -> bool {
+        self.sink.is_terminated()
+    }
+}

futures-util/src/sink/map_err.rs

@@ -1,5 +1,5 @@
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use futures_sink::{Sink};
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
@@ -85,6 +85,7 @@ impl<Si, F, E, Item> Sink<Item> for SinkMapErr<Si, F>
     }
 }
 
+// Forwarding impl of Stream from the underlying sink
 impl<S: Stream, F> Stream for SinkMapErr<S, F> {
     type Item = S::Item;
 
@@ -95,3 +96,9 @@ impl<S: Stream, F> Stream for SinkMapErr<S, F> {
         self.sink().poll_next(cx)
     }
 }
+
+impl<S: FusedStream, F> FusedStream for SinkMapErr<S, F> {
+    fn is_terminated(&self) -> bool {
+        self.sink.is_terminated()
+    }
+}

futures-util/src/sink/with_flat_map.rs

@@ -1,7 +1,7 @@
 use core::fmt;
 use core::marker::PhantomData;
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
@@ -113,6 +113,7 @@ where
     }
 }
 
+// Forwarding impl of Stream from the underlying sink
 impl<S, Item, U, St, F> Stream for WithFlatMap<S, Item, U, St, F>
 where
     S: Stream + Sink<Item>,
@@ -128,6 +129,17 @@ where
     }
 }
 
+impl<S, Item, U, St, F> FusedStream for WithFlatMap<S, Item, U, St, F>
+where
+    S: FusedStream + Sink<Item>,
+    F: FnMut(U) -> St,
+    St: Stream<Item = Result<Item, S::Error>>,
+{
+    fn is_terminated(&self) -> bool {
+        self.sink.is_terminated()
+    }
+}
+
 impl<Si, Item, U, St, F> Sink<U> for WithFlatMap<Si, Item, U, St, F>
 where
     Si: Sink<Item>,

futures-util/src/stream/catch_unwind.rs

@@ -1,4 +1,4 @@
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 use std::any::Any;
@@ -22,8 +22,7 @@ impl<St: Stream + UnwindSafe> CatchUnwind<St> {
     }
 }
 
-impl<St: Stream + UnwindSafe> Stream for CatchUnwind<St>
-{
+impl<St: Stream + UnwindSafe> Stream for CatchUnwind<St> {
     type Item = Result<St::Item, Box<dyn Any + Send>>;
 
     fn poll_next(
@@ -47,3 +46,9 @@ impl<St: Stream + UnwindSafe> Stream for CatchUnwind<St>
         }
     }
 }
+
+impl<St: FusedStream + UnwindSafe> FusedStream for CatchUnwind<St> {
+    fn is_terminated(&self) -> bool {
+        self.caught_unwind || self.stream.is_terminated()
+    }
+}

futures-util/src/stream/chunks.rs

@@ -1,5 +1,5 @@
 use crate::stream::Fuse;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -107,6 +107,12 @@ impl<St: Stream> Stream for Chunks<St> {
     }
 }
 
+impl<St: FusedStream> FusedStream for Chunks<St> {
+    fn is_terminated(&self) -> bool {
+        self.stream.is_terminated() && self.items.is_empty()
+    }
+}
+
 // Forwarding impl of Sink from the underlying stream
 #[cfg(feature = "sink")]
 impl<S, Item> Sink<Item> for Chunks<S>

futures-util/src/stream/once.rs

@@ -1,6 +1,6 @@
 use core::pin::Pin;
 use futures_core::future::Future;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use pin_utils::unsafe_pinned;
 
@@ -51,3 +51,9 @@ impl<Fut: Future> Stream for Once<Fut> {
         Poll::Ready(Some(val))
     }
 }
+
+impl<Fut: Future> FusedStream for Once<Fut> {
+    fn is_terminated(&self) -> bool {
+        self.future.is_none()
+    }
+}

futures-util/src/stream/repeat.rs

@@ -1,5 +1,5 @@
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 
 /// Stream for the [`repeat`] function.
@@ -40,3 +40,11 @@ impl<T> Stream for Repeat<T>
         Poll::Ready(Some(self.item.clone()))
     }
 }
+
+impl<T> FusedStream for Repeat<T>
+    where T: Clone,
+{
+    fn is_terminated(&self) -> bool {
+        false
+    }
+}

futures-util/src/stream/skip_while.rs

@@ -95,7 +95,7 @@ impl<St, Fut, F> FusedStream for SkipWhile<St, Fut, F>
           Fut: Future<Output = bool>,
 {
     fn is_terminated(&self) -> bool {
-        self.stream.is_terminated()
+        self.pending_item.is_none() && self.stream.is_terminated()
     }
 }
 

futures-util/src/stream/take.rs

@@ -1,5 +1,5 @@
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -81,6 +81,14 @@ impl<St> Stream for Take<St>
     }
 }
 
+impl<St> FusedStream for Take<St>
+    where St: FusedStream,
+{
+    fn is_terminated(&self) -> bool {
+        self.remaining == 0 || self.stream.is_terminated()
+    }
+}
+
 // Forwarding impl of Sink from the underlying stream
 #[cfg(feature = "sink")]
 impl<S, Item> Sink<Item> for Take<S>

futures-util/src/stream/take_while.rs

@@ -1,7 +1,7 @@
 use core::fmt;
 use core::pin::Pin;
 use futures_core::future::Future;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -129,6 +129,16 @@ impl<St, Fut, F> Stream for TakeWhile<St, Fut, F>
     }
 }
 
+impl<St, Fut, F> FusedStream for TakeWhile<St, Fut, F>
+    where St: FusedStream,
+          F: FnMut(&St::Item) -> Fut,
+          Fut: Future<Output = bool>,
+{
+    fn is_terminated(&self) -> bool {
+        self.done_taking || self.pending_item.is_none() && self.stream.is_terminated()
+    }
+}
+
 // Forwarding impl of Sink from the underlying stream
 #[cfg(feature = "sink")]
 impl<S, Fut, F, Item> Sink<Item> for TakeWhile<S, Fut, F>

futures-util/src/try_stream/and_then.rs

@@ -1,7 +1,7 @@
 use core::fmt;
 use core::pin::Pin;
 use futures_core::future::TryFuture;
-use futures_core::stream::{Stream, TryStream};
+use futures_core::stream::{Stream, TryStream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -104,6 +104,16 @@ impl<St, Fut, F> Stream for AndThen<St, Fut, F>
     }
 }
 
+impl<St, Fut, F> FusedStream for AndThen<St, Fut, F>
+    where St: TryStream + FusedStream,
+          F: FnMut(St::Ok) -> Fut,
+          Fut: TryFuture<Error = St::Error>,
+{
+    fn is_terminated(&self) -> bool {
+        self.future.is_none() && self.stream.is_terminated()
+    }
+}
+
 // Forwarding impl of Sink from the underlying stream
 #[cfg(feature = "sink")]
 impl<S, Fut, F, Item> Sink<Item> for AndThen<S, Fut, F>

futures-util/src/try_stream/or_else.rs

@@ -1,7 +1,7 @@
 use core::fmt;
 use core::pin::Pin;
 use futures_core::future::TryFuture;
-use futures_core::stream::{Stream, TryStream};
+use futures_core::stream::{Stream, TryStream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -105,6 +105,16 @@ impl<St, Fut, F> Stream for OrElse<St, Fut, F>
     }
 }
 
+impl<St, Fut, F> FusedStream for OrElse<St, Fut, F>
+    where St: TryStream + FusedStream,
+          F: FnMut(St::Error) -> Fut,
+          Fut: TryFuture<Ok = St::Ok>,
+{
+    fn is_terminated(&self) -> bool {
+        self.future.is_none() && self.stream.is_terminated()
+    }
+}
+
 // Forwarding impl of Sink from the underlying stream
 #[cfg(feature = "sink")]
 impl<S, Fut, F, Item> Sink<Item> for OrElse<S, Fut, F>

futures-util/src/try_stream/try_skip_while.rs

@@ -1,7 +1,7 @@
 use core::fmt;
 use core::pin::Pin;
 use futures_core::future::TryFuture;
-use futures_core::stream::{Stream, TryStream};
+use futures_core::stream::{Stream, TryStream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -133,6 +133,16 @@ impl<St, Fut, F> Stream for TrySkipWhile<St, Fut, F>
     }
 }
 
+impl<St, Fut, F> FusedStream for TrySkipWhile<St, Fut, F>
+    where St: TryStream + FusedStream,
+          F: FnMut(&St::Ok) -> Fut,
+          Fut: TryFuture<Ok = bool, Error = St::Error>,
+{
+    fn is_terminated(&self) -> bool {
+        self.pending_item.is_none() && self.stream.is_terminated()
+    }
+}
+
 // Forwarding impl of Sink from the underlying stream
 #[cfg(feature = "sink")]
 impl<S, Fut, F, Item, E> Sink<Item> for TrySkipWhile<S, Fut, F>