Add FusedFuture implementations

futures-test/src/future/pending_once.rs

@@ -45,7 +45,7 @@ impl<Fut: Future> Future for PendingOnce<Fut> {
     }
 }
 
-impl<Fut: Future + FusedFuture> FusedFuture for PendingOnce<Fut> {
+impl<Fut: FusedFuture> FusedFuture for PendingOnce<Fut> {
     fn is_terminated(&self) -> bool {
         self.polled_before && self.future.is_terminated()
     }

futures-test/src/interleave_pending.rs

@@ -1,5 +1,5 @@
-use futures_core::future::Future;
-use futures_core::stream::Stream;
+use futures_core::future::{Future, FusedFuture};
+use futures_core::stream::{Stream, FusedStream};
 use futures_io::{self as io, AsyncBufRead, AsyncRead, AsyncWrite};
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 use std::{
@@ -85,6 +85,12 @@ impl<Fut: Future> Future for InterleavePending<Fut> {
     }
 }
 
+impl<Fut: FusedFuture> FusedFuture for InterleavePending<Fut> {
+    fn is_terminated(&self) -> bool {
+        self.inner.is_terminated()
+    }
+}
+
 impl<St: Stream> Stream for InterleavePending<St> {
     type Item = St::Item;
 
@@ -110,6 +116,12 @@ impl<St: Stream> Stream for InterleavePending<St> {
     }
 }
 
+impl<Fut: FusedStream> FusedStream for InterleavePending<Fut> {
+    fn is_terminated(&self) -> bool {
+        self.inner.is_terminated()
+    }
+}
+
 impl<W: AsyncWrite> AsyncWrite for InterleavePending<W> {
     fn poll_write(
         self: Pin<&mut Self>,

futures-util/src/future/into_stream.rs

@@ -1,7 +1,7 @@
 use crate::stream::{self, Once};
 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;
 
@@ -35,3 +35,9 @@ impl<Fut: Future> Stream for IntoStream<Fut> {
         self.inner.size_hint()
     }
 }
+
+impl<Fut: Future> FusedStream for IntoStream<Fut> {
+    fn is_terminated(&self) -> bool {
+        self.inner.is_terminated()
+    }
+}

futures-util/src/future/join.rs

@@ -3,7 +3,7 @@
 use crate::future::{MaybeDone, maybe_done};
 use core::fmt;
 use core::pin::Pin;
-use futures_core::future::Future;
+use futures_core::future::{Future, FusedFuture};
 use futures_core::task::{Context, Poll};
 use pin_utils::unsafe_pinned;
 use super::assert_future;
@@ -62,6 +62,14 @@ macro_rules! generate {
                 }
             }
         }
+
+        impl<$($Fut: FusedFuture),*> FusedFuture for $Join<$($Fut),*> {
+            fn is_terminated(&self) -> bool {
+                $(
+                    self.$Fut.is_terminated()
+                ) && *
+            }
+        }
     )*)
 }
 

futures-util/src/future/option.rs

@@ -1,7 +1,7 @@
 //! Definition of the `Option` (optional step) combinator
 
 use core::pin::Pin;
-use futures_core::future::Future;
+use futures_core::future::{Future, FusedFuture};
 use futures_core::task::{Context, Poll};
 use pin_utils::unsafe_pinned;
 
@@ -46,6 +46,15 @@ impl<F: Future> Future for OptionFuture<F> {
     }
 }
 
+impl<F: FusedFuture> FusedFuture for OptionFuture<F> {
+    fn is_terminated(&self) -> bool {
+        match &self.option {
+            Some(x) => x.is_terminated(),
+            None => true,
+        }
+    }
+}
+
 impl<T> From<Option<T>> for OptionFuture<T> {
     fn from(option: Option<T>) -> Self {
         OptionFuture { option }

futures-util/src/future/select.rs

@@ -1,12 +1,12 @@
 use core::pin::Pin;
-use futures_core::future::Future;
+use futures_core::future::{Future, FusedFuture};
 use futures_core::task::{Context, Poll};
 use crate::future::{Either, FutureExt};
 
 /// Future for the [`select()`] function.
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 #[derive(Debug)]
-pub struct Select<A: Unpin, B: Unpin> {
+pub struct Select<A, B> {
     inner: Option<(A, B)>,
 }
 
@@ -50,7 +50,11 @@ pub fn select<A, B>(future1: A, future2: B) -> Select<A, B>
     Select { inner: Some((future1, future2)) }
 }
 
-impl<A: Unpin, B: Unpin> Future for Select<A, B> where A: Future, B: Future {
+impl<A, B> Future for Select<A, B>
+where
+    A: Future + Unpin,
+    B: Future + Unpin,
+{
     type Output = Either<(A::Output, B), (B::Output, A)>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
@@ -67,3 +71,13 @@ impl<A: Unpin, B: Unpin> Future for Select<A, B> where A: Future, B: Future {
         }
     }
 }
+
+impl<A, B> FusedFuture for Select<A, B>
+where
+    A: Future + Unpin,
+    B: Future + Unpin,
+{
+    fn is_terminated(&self) -> bool {
+        self.inner.is_none()
+    }
+}

futures-util/src/stream/concat.rs

@@ -1,6 +1,6 @@
 use core::pin::Pin;
-use futures_core::future::Future;
-use futures_core::stream::Stream;
+use futures_core::future::{Future, FusedFuture};
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
@@ -57,3 +57,13 @@ where St: Stream,
         }
     }
 }
+
+impl<St> FusedFuture for Concat<St>
+where St: FusedStream,
+      St::Item: Extend<<St::Item as IntoIterator>::Item> +
+                IntoIterator + Default,
+{
+    fn is_terminated(&self) -> bool {
+        self.accum.is_none() && self.stream.is_terminated()
+    }
+}