Add Future/Stream bounds to FusedFuture/FusedStream

futures-core/src/future/mod.rs

@@ -18,20 +18,20 @@ pub type BoxFuture<'a, T> = Pin<alloc::boxed::Box<dyn Future<Output = T> + Send
 #[cfg(feature = "alloc")]
 pub type LocalBoxFuture<'a, T> = Pin<alloc::boxed::Box<dyn Future<Output = T> + 'a>>;
 
-/// A `Future` or `TryFuture` which tracks whether or not the underlying future
+/// A future which tracks whether or not the underlying future
 /// should no longer be polled.
 ///
 /// `is_terminated` will return `true` if a future should no longer be polled.
 /// Usually, this state occurs after `poll` (or `try_poll`) returned
 /// `Poll::Ready`. However, `is_terminated` may also return `true` if a future
 /// has become inactive and can no longer make progress and should be ignored
 /// or dropped rather than being `poll`ed again.
-pub trait FusedFuture {
+pub trait FusedFuture: Future {
     /// Returns `true` if the underlying future should no longer be polled.
     fn is_terminated(&self) -> bool;
 }
 
-impl<F: FusedFuture + ?Sized> FusedFuture for &mut F {
+impl<F: FusedFuture + ?Sized + Unpin> FusedFuture for &mut F {
     fn is_terminated(&self) -> bool {
         <F as FusedFuture>::is_terminated(&**self)
     }
@@ -92,7 +92,7 @@ mod if_alloc {
     use alloc::boxed::Box;
     use super::*;
 
-    impl<F: FusedFuture + ?Sized> FusedFuture for Box<F> {
+    impl<F: FusedFuture + ?Sized + Unpin> FusedFuture for Box<F> {
         fn is_terminated(&self) -> bool {
             <F as FusedFuture>::is_terminated(&**self)
         }

futures-core/src/stream.rs

@@ -87,20 +87,20 @@ where
     }
 }
 
-/// A `Stream` or `TryStream` which tracks whether or not the underlying stream
+/// A stream which tracks whether or not the underlying stream
 /// should no longer be polled.
 ///
 /// `is_terminated` will return `true` if a future should no longer be polled.
 /// Usually, this state occurs after `poll_next` (or `try_poll_next`) returned
 /// `Poll::Ready(None)`. However, `is_terminated` may also return `true` if a
 /// stream has become inactive and can no longer make progress and should be
 /// ignored or dropped rather than being polled again.
-pub trait FusedStream {
+pub trait FusedStream: Stream {
     /// Returns `true` if the stream should no longer be polled.
     fn is_terminated(&self) -> bool;
 }
 
-impl<F: ?Sized + FusedStream> FusedStream for &mut F {
+impl<F: ?Sized + FusedStream + Unpin> FusedStream for &mut F {
     fn is_terminated(&self) -> bool {
         <F as FusedStream>::is_terminated(&**self)
     }
@@ -194,7 +194,7 @@ mod if_alloc {
         }
     }
 
-    impl<S: ?Sized + FusedStream> FusedStream for Box<S> {
+    impl<S: ?Sized + FusedStream + Unpin> FusedStream for Box<S> {
         fn is_terminated(&self) -> bool {
             <S as FusedStream>::is_terminated(&**self)
         }

futures-util/src/future/either.rs

@@ -69,7 +69,7 @@ where
 impl<A, B> FusedFuture for Either<A, B>
 where
     A: FusedFuture,
-    B: FusedFuture,
+    B: FusedFuture<Output = A::Output>,
 {
     fn is_terminated(&self) -> bool {
         match self {
@@ -99,7 +99,7 @@ where
 impl<A, B> FusedStream for Either<A, B>
 where
     A: FusedStream,
-    B: FusedStream,
+    B: FusedStream<Item = A::Item>,
 {
     fn is_terminated(&self) -> bool {
         match self {

futures-util/src/future/inspect.rs

@@ -25,7 +25,10 @@ impl<Fut: Future, F: FnOnce(&Fut::Output)> Inspect<Fut, F> {
 
 impl<Fut: Future + Unpin, F> Unpin for Inspect<Fut, F> {}
 
-impl<Fut: Future + FusedFuture, F> FusedFuture for Inspect<Fut, F> {
+impl<Fut, F> FusedFuture for Inspect<Fut, F>
+    where Fut: FusedFuture,
+          F: FnOnce(&Fut::Output),
+{
     fn is_terminated(&self) -> bool { self.future.is_terminated() }
 }
 

futures-util/src/future/lazy.rs

@@ -38,11 +38,13 @@ pub fn lazy<F, R>(f: F) -> Lazy<F>
     Lazy { f: Some(f) }
 }
 
-impl<F> FusedFuture for Lazy<F> {
+impl<F, R> FusedFuture for Lazy<F>
+    where F: FnOnce(&mut Context<'_>) -> R,
+{
     fn is_terminated(&self) -> bool { self.f.is_none() }
 }
 
-impl<R, F> Future for Lazy<F>
+impl<F, R> Future for Lazy<F>
     where F: FnOnce(&mut Context<'_>) -> R,
 {
     type Output = R;

futures-util/src/future/map.rs

@@ -23,7 +23,10 @@ impl<Fut, F> Map<Fut, F> {
 
 impl<Fut: Unpin, F> Unpin for Map<Fut, F> {}
 
-impl<Fut, F> FusedFuture for Map<Fut, F> {
+impl<Fut, F, T> FusedFuture for Map<Fut, F>
+    where Fut: Future,
+          F: FnOnce(Fut::Output) -> T,
+{
     fn is_terminated(&self) -> bool { self.f.is_none() }
 }
 

futures-util/src/future/shared.rs

@@ -173,14 +173,19 @@ where
     }
 }
 
-impl<Fut: Future> FusedFuture for Shared<Fut> {
+impl<Fut> FusedFuture for Shared<Fut>
+where
+    Fut: Future,
+    Fut::Output: Clone,
+{
     fn is_terminated(&self) -> bool {
         self.inner.is_none()
     }
 }
 
-impl<Fut: Future> Future for Shared<Fut>
+impl<Fut> Future for Shared<Fut>
 where
+    Fut: Future,
     Fut::Output: Clone,
 {
     type Output = Fut::Output;

futures-util/src/future/then.rs

@@ -25,7 +25,11 @@ impl<Fut1, Fut2, F> Then<Fut1, Fut2, F>
     }
 }
 
-impl<Fut1, Fut2, F> FusedFuture for Then<Fut1, Fut2, F> {
+impl<Fut1, Fut2, F> FusedFuture for Then<Fut1, Fut2, F>
+    where Fut1: Future,
+          Fut2: Future,
+          F: FnOnce(Fut1::Output) -> Fut2,
+{
     fn is_terminated(&self) -> bool { self.chain.is_terminated() }
 }
 

futures-util/src/stream/chain.rs

@@ -27,7 +27,10 @@ where St1: Stream,
     }
 }
 
-impl<St1, St2: FusedStream> FusedStream for Chain<St1, St2> {
+impl<St1, St2> FusedStream for Chain<St1, St2>
+where St1: Stream,
+      St2: FusedStream<Item=St1::Item>,
+{
     fn is_terminated(&self) -> bool {
         self.first.is_none() && self.second.is_terminated()
     }

futures-util/src/stream/collect.rs

@@ -31,7 +31,10 @@ impl<St: Stream, C: Default> Collect<St, C> {
     }
 }
 
-impl<St: FusedStream, C> FusedFuture for Collect<St, C> {
+impl<St, C> FusedFuture for Collect<St, C>
+where St: FusedStream,
+      C: Default + Extend<St:: Item>
+{
     fn is_terminated(&self) -> bool {
         self.stream.is_terminated()
     }

futures-util/src/stream/flatten.rs

@@ -75,7 +75,8 @@ where
 
 impl<St> FusedStream for Flatten<St>
 where
-    St: Stream + FusedStream,
+    St: FusedStream,
+    St::Item: Stream,
 {
     fn is_terminated(&self) -> bool {
         self.next.is_none() && self.stream.is_terminated()

futures-util/src/stream/fold.rs

@@ -51,7 +51,11 @@ where St: Stream,
     }
 }
 
-impl<St, Fut, T, F> FusedFuture for Fold<St, Fut, T, F> {
+impl<St, Fut, T, F> FusedFuture for Fold<St, Fut, T, F>
+    where St: Stream,
+          F: FnMut(T, St::Item) -> Fut,
+          Fut: Future<Output = T>,
+{
     fn is_terminated(&self) -> bool {
         self.accum.is_none() && self.future.is_none()
     }

futures-util/src/stream/for_each.rs

@@ -50,7 +50,11 @@ where St: Stream,
     }
 }
 
-impl<St: FusedStream, Fut, F> FusedFuture for ForEach<St, Fut, F> {
+impl<St, Fut, F> FusedFuture for ForEach<St, Fut, F>
+    where St: FusedStream,
+          F: FnMut(St::Item) -> Fut,
+          Fut: Future<Output = ()>,
+{
     fn is_terminated(&self) -> bool {
         self.future.is_none() && self.stream.is_terminated()
     }

futures-util/src/stream/for_each_concurrent.rs

@@ -57,7 +57,11 @@ where St: Stream,
     }
 }
 
-impl<St, Fut, F> FusedFuture for ForEachConcurrent<St, Fut, F> {
+impl<St, Fut, F> FusedFuture for ForEachConcurrent<St, Fut, F>
+    where St: Stream,
+          F: FnMut(St::Item) -> Fut,
+          Fut: Future<Output = ()>,
+{
     fn is_terminated(&self) -> bool {
         self.stream.is_none() && self.futures.is_empty()
     }

futures-util/src/stream/forward.rs

@@ -53,7 +53,11 @@ where
     }
 }
 
-impl<St: TryStream, Si: Sink<St::Ok> + Unpin> FusedFuture for Forward<St, Si> {
+impl<St, Si, Item, E> FusedFuture for Forward<St, Si>
+where
+    Si: Sink<Item, Error = E>,
+    St: Stream<Item = Result<Item, E>>,
+{
     fn is_terminated(&self) -> bool {
         self.sink.is_none()
     }

futures-util/src/stream/fuse.rs

@@ -65,7 +65,7 @@ impl<St> Fuse<St> {
     }
 }
 
-impl<S> FusedStream for Fuse<S> {
+impl<S: Stream> FusedStream for Fuse<S> {
     fn is_terminated(&self) -> bool {
         self.done
     }

futures-util/src/stream/inspect.rs

@@ -70,7 +70,10 @@ impl<St, F> Inspect<St, F>
     }
 }
 
-impl<St: Stream + FusedStream, F> FusedStream for Inspect<St, F> {
+impl<St, F> FusedStream for Inspect<St, F>
+    where St: FusedStream,
+          F: FnMut(&St::Item),
+{
     fn is_terminated(&self) -> bool {
         self.stream.is_terminated()
     }

futures-util/src/stream/into_future.rs

@@ -71,7 +71,7 @@ impl<St: Stream + Unpin> StreamFuture<St> {
     }
 }
 
-impl<St> FusedFuture for StreamFuture<St> {
+impl<St: Stream + Unpin> FusedFuture for StreamFuture<St> {
     fn is_terminated(&self) -> bool {
         self.stream.is_none()
     }

futures-util/src/stream/map.rs

@@ -70,7 +70,10 @@ impl<St, T, F> Map<St, F>
     }
 }
 
-impl<St: FusedStream, F> FusedStream for Map<St, F> {
+impl<St, F, T> FusedStream for Map<St, F>
+    where St: FusedStream,
+          F: FnMut(St::Item) -> T,
+{
     fn is_terminated(&self) -> bool {
         self.stream.is_terminated()
     }

futures-util/src/stream/next.rs

@@ -19,7 +19,7 @@ impl<'a, St: ?Sized + Stream + Unpin> Next<'a, St> {
     }
 }
 
-impl<St: ?Sized + FusedStream> FusedFuture for Next<'_, St> {
+impl<St: ?Sized + FusedStream + Unpin> FusedFuture for Next<'_, St> {
     fn is_terminated(&self) -> bool {
         self.stream.is_terminated()
     }

futures-util/src/stream/select.rs

@@ -72,7 +72,10 @@ impl<St1, St2> Select<St1, St2> {
     }
 }
 
-impl<St1, St2> FusedStream for Select<St1, St2> {
+impl<St1, St2> FusedStream for Select<St1, St2>
+    where St1: Stream,
+          St2: Stream<Item = St1::Item>
+{
     fn is_terminated(&self) -> bool {
         self.stream1.is_terminated() && self.stream2.is_terminated()
     }

futures-util/src/stream/select_next_some.rs

@@ -1,5 +1,5 @@
 use core::pin::Pin;
-use futures_core::stream::{Stream, FusedStream};
+use futures_core::stream::FusedStream;
 use futures_core::future::{Future, FusedFuture};
 use futures_core::task::{Context, Poll};
 use crate::stream::StreamExt;
@@ -18,13 +18,13 @@ impl<'a, St: ?Sized> SelectNextSome<'a, St> {
     }
 }
 
-impl<St: ?Sized + FusedStream> FusedFuture for SelectNextSome<'_, St> {
+impl<St: ?Sized + FusedStream + Unpin> FusedFuture for SelectNextSome<'_, St> {
     fn is_terminated(&self) -> bool {
         self.stream.is_terminated()
     }
 }
 
-impl<St: ?Sized + Stream + FusedStream + Unpin> Future for SelectNextSome<'_, St> {
+impl<St: ?Sized + FusedStream + Unpin> Future for SelectNextSome<'_, St> {
     type Output = St::Item;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {

futures-util/src/stream/skip_while.rs

@@ -89,7 +89,11 @@ impl<St, Fut, F> SkipWhile<St, Fut, F>
     }
 }
 
-impl<St: Stream + FusedStream, Fut, F> FusedStream for SkipWhile<St, Fut, F> {
+impl<St, Fut, F> FusedStream for SkipWhile<St, Fut, F>
+    where St: FusedStream,
+          F: FnMut(&St::Item) -> Fut,
+          Fut: Future<Output = bool>,
+{
     fn is_terminated(&self) -> bool {
         self.stream.is_terminated()
     }

futures-util/src/stream/then.rs

@@ -81,7 +81,11 @@ impl<St, Fut, F> Then<St, Fut, F>
     }
 }
 
-impl<St: FusedStream, Fut, F> FusedStream for Then<St, Fut, F> {
+impl<St, Fut, F> FusedStream for Then<St, Fut, F>
+    where St: FusedStream,
+          F: FnMut(St::Item) -> Fut,
+          Fut: Future,
+{
     fn is_terminated(&self) -> bool {
         self.future.is_none() && self.stream.is_terminated()
     }

futures-util/src/stream/unfold.rs

@@ -50,9 +50,9 @@ use pin_utils::{unsafe_pinned, unsafe_unpinned};
 /// assert_eq!(result, vec![0, 2, 4]);
 /// # });
 /// ```
-pub fn unfold<T, F, Fut, It>(init: T, f: F) -> Unfold<T, F, Fut>
+pub fn unfold<T, F, Fut, Item>(init: T, f: F) -> Unfold<T, F, Fut>
     where F: FnMut(T) -> Fut,
-          Fut: Future<Output = Option<(It, T)>>,
+          Fut: Future<Output = Option<(Item, T)>>,
 {
     Unfold {
         f,
@@ -90,22 +90,25 @@ impl<T, F, Fut> Unfold<T, F, Fut> {
     unsafe_pinned!(fut: Option<Fut>);
 }
 
-impl<T, F, Fut> FusedStream for Unfold<T, F, Fut> {
+impl<T, F, Fut, Item> FusedStream for Unfold<T, F, Fut>
+    where F: FnMut(T) -> Fut,
+          Fut: Future<Output = Option<(Item, T)>>,
+{
     fn is_terminated(&self) -> bool {
         self.state.is_none() && self.fut.is_none()
     }
 }
 
-impl<T, F, Fut, It> Stream for Unfold<T, F, Fut>
+impl<T, F, Fut, Item> Stream for Unfold<T, F, Fut>
     where F: FnMut(T) -> Fut,
-          Fut: Future<Output = Option<(It, T)>>,
+          Fut: Future<Output = Option<(Item, T)>>,
 {
-    type Item = It;
+    type Item = Item;
 
     fn poll_next(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<It>> {
+    ) -> Poll<Option<Self::Item>> {
         if let Some(state) = self.as_mut().state().take() {
             let fut = (self.as_mut().f())(state);
             self.as_mut().fut().set(Some(fut));