Merge pull request rust-lang#45 from Stebalien/fix-stream-poll

Make stream::Fuse return None after finishing, not NotReady

src/stream/fuse.rs

@@ -21,11 +21,10 @@ impl<S: Stream> Stream for Fuse<S> {
 
     fn poll(&mut self, task: &mut Task) -> Poll<Option<S::Item>, S::Error> {
         let ret = self.stream.as_mut().map(|s| s.poll(task));
-        match ret {
-            Some(Poll::Ok(None)) => self.stream = None,
-            _ => {}
+        if let Some(Poll::Ok(None)) = ret {
+            self.stream = None;
         }
-        ret.unwrap_or(Poll::NotReady)
+        ret.unwrap_or(Poll::Ok(None))
     }
 
     fn schedule(&mut self, task: &mut Task) {

tests/stream.rs

@@ -1,6 +1,6 @@
 extern crate futures;
 
-use futures::{failed, finished, Future, promise};
+use futures::{failed, finished, Future, promise, Poll, Task};
 use futures::stream::*;
 
 mod support;
@@ -77,6 +77,18 @@ fn adapters() {
     assert_done(|| list().skip(2).collect(), Ok(vec![3]));
 }
 
+#[test]
+fn fuse() {
+    let mut stream = list().fuse();
+    let mut task = Task::new();
+    assert_eq!(stream.poll(&mut task), Poll::Ok(Some(1)));
+    assert_eq!(stream.poll(&mut task), Poll::Ok(Some(2)));
+    assert_eq!(stream.poll(&mut task), Poll::Ok(Some(3)));
+    assert_eq!(stream.poll(&mut task), Poll::Ok(None));
+    assert_eq!(stream.poll(&mut task), Poll::Ok(None));
+    assert_eq!(stream.poll(&mut task), Poll::Ok(None));
+}
+
 // #[test]
 // fn adapters_poll() {
 //     assert_eq!(collect_poll(list().map(|a| a + 1)), Ok(vec![2, 3, 4]));