Test non-obvious fused behaviour

FuturesUnordered can reset from being terminated when it receives new
tasks to process. SelectNextSome must handle this to ensure the example
pattern of pushing tasks in a looped `select!` works.

Author: Nemo157

futures-util/tests/futures_unordered.rs

@@ -0,0 +1,24 @@
+#![feature(async_await, await_macro, futures_api, pin)]
+
+use futures::future;
+use futures::task::Poll;
+use futures::stream::{FusedStream, FuturesUnordered, StreamExt};
+use futures_test::task::noop_local_waker_ref;
+
+#[test]
+fn is_terminated() {
+    let lw = noop_local_waker_ref();
+    let mut tasks = FuturesUnordered::new();
+
+    assert_eq!(tasks.is_terminated(), false);
+    assert_eq!(tasks.poll_next_unpin(lw), Poll::Ready(None));
+    assert_eq!(tasks.is_terminated(), true);
+
+    tasks.push(future::ready(1));
+
+    assert_eq!(tasks.is_terminated(), false);
+    assert_eq!(tasks.poll_next_unpin(lw), Poll::Ready(Some(1)));
+    assert_eq!(tasks.is_terminated(), false);
+    assert_eq!(tasks.poll_next_unpin(lw), Poll::Ready(None));
+    assert_eq!(tasks.is_terminated(), true);
+}

futures-util/tests/select_next_some.rs

@@ -1,12 +1,37 @@
 #![feature(async_await, await_macro, futures_api, pin)]
 
 use futures::{future, select};
-use futures::stream::FuturesUnordered;
-use futures::stream::StreamExt;
+use futures::future::{FusedFuture, FutureExt};
+use futures::stream::{FuturesUnordered, StreamExt};
+use futures::task::Poll;
 use futures_test::future::FutureTestExt;
+use futures_test::task::WakeCounter;
 
 #[test]
-fn futures_unordered() {
+fn is_terminated() {
+    let counter = WakeCounter::new();
+
+    let mut tasks = FuturesUnordered::new();
+
+    let mut select_next_some = tasks.select_next_some();
+    assert_eq!(select_next_some.is_terminated(), false);
+    assert_eq!(select_next_some.poll_unpin(counter.local_waker()), Poll::Pending);
+    assert_eq!(counter.count(), 1);
+    assert_eq!(select_next_some.is_terminated(), true);
+    drop(select_next_some);
+
+    tasks.push(future::ready(1));
+
+    let mut select_next_some = tasks.select_next_some();
+    assert_eq!(select_next_some.is_terminated(), false);
+    assert_eq!(select_next_some.poll_unpin(counter.local_waker()), Poll::Ready(1));
+    assert_eq!(select_next_some.is_terminated(), false);
+    assert_eq!(select_next_some.poll_unpin(counter.local_waker()), Poll::Pending);
+    assert_eq!(select_next_some.is_terminated(), true);
+}
+
+#[test]
+fn select() {
     // Checks that even though `async_tasks` will yield a `None` and return
     // `is_terminated() == true` during the first poll, it manages to toggle
     // back to having items after a future is pushed into it during the second

futures/src/lib.rs

@@ -174,7 +174,7 @@ pub mod future {
     //!   immediate defined value.
 
     pub use futures_core::future::{
-        Future, TryFuture,
+        Future, TryFuture, FusedFuture,
         FutureObj, LocalFutureObj, UnsafeFutureObj,
     };
 
@@ -295,8 +295,8 @@ pub mod stream {
     //!   constructs a stream from a collection of futures.
 
     pub use futures_core::stream::{
-        Stream, TryStream,
-        StreamObj, LocalStreamObj, UnsafeStreamObj
+        Stream, TryStream, FusedStream,
+        StreamObj, LocalStreamObj, UnsafeStreamObj,
     };
 
     pub use futures_util::stream::{