Revert "Generalize "unfold" combinator."

This reverts commit 84a4cd0.

Author: cramertj

src/stream/unfold.rs

@@ -13,7 +13,7 @@ use stream::Stream;
 /// for the returned `Future` to complete with `(a, b)`. It will then yield the
 /// value `a`, and use `b` as the next internal state.
 ///
-/// If the future returns `None` instead of `Some`, then the `unfold()`
+/// If the closure returns `None` instead of `Some(Future)`, then the `unfold()`
 /// will stop producing items and return `Ok(Async::Ready(None))` in future
 /// calls to `poll()`.
 ///
@@ -33,23 +33,22 @@ use stream::Stream;
 /// use futures::future::{self, Future};
 ///
 /// let mut stream = stream::unfold(0, |state| {
-///     future::ok::<_, u32>(
-///         if state <= 2 {
-///             let next_state = state + 1;
-///             let yielded = state  * 2;
-///             Some((yielded, next_state))
-///         } else {
-///             None
-///         }
-///     )
+///     if state <= 2 {
+///         let next_state = state + 1;
+///         let yielded = state  * 2;
+///         let fut = future::ok::<_, u32>((yielded, next_state));
+///         Some(fut)
+///     } else {
+///         None
+///     }
 /// });
 ///
 /// let result = stream.collect().wait();
 /// assert_eq!(result, Ok(vec![0, 2, 4]));
 /// ```
 pub fn unfold<T, F, Fut, It>(init: T, f: F) -> Unfold<T, F, Fut>
-    where F: FnMut(T) -> Fut,
-          Fut: IntoFuture<Item = Option<(It, T)>>,
+    where F: FnMut(T) -> Option<Fut>,
+          Fut: IntoFuture<Item = (It, T)>,
 {
     Unfold {
         f: f,
@@ -68,8 +67,8 @@ pub struct Unfold<T, F, Fut> where Fut: IntoFuture {
 }
 
 impl <T, F, Fut, It> Stream for Unfold<T, F, Fut>
-    where F: FnMut(T) -> Fut,
-          Fut: IntoFuture<Item = Option<(It, T)>>,
+    where F: FnMut(T) -> Option<Fut>,
+          Fut: IntoFuture<Item = (It, T)>,
 {
     type Item = It;
     type Error = Fut::Error;
@@ -80,18 +79,17 @@ impl <T, F, Fut, It> Stream for Unfold<T, F, Fut>
                 // State::Empty may happen if the future returned an error
                 State::Empty => { return Ok(Async::Ready(None)); }
                 State::Ready(state) => {
-                    let fut = (self.f)(state);
-                    self.state = State::Processing(fut.into_future());
+                    match (self.f)(state) {
+                        Some(fut) => { self.state = State::Processing(fut.into_future()); }
+                        None => { return Ok(Async::Ready(None)); }
+                    }
                 }
                 State::Processing(mut fut) => {
                     match fut.poll()? {
-                        Async::Ready(Some((item, next_state))) => {
+                        Async:: Ready((item, next_state)) => {
                             self.state = State::Ready(next_state);
                             return Ok(Async::Ready(Some(item)));
                         }
-                        Async::Ready(None) => {
-                            return Ok(Async::Ready(None));
-                        }
                         Async::NotReady => {
                             self.state = State::Processing(fut);
                             return Ok(Async::NotReady);

tests/unfold.rs

@@ -3,18 +3,17 @@ extern crate futures;
 mod support;
 
 use futures::stream;
-use futures::future::{ok, Either};
 
 use support::*;
 
 #[test]
 fn unfold1() {
     let mut stream = stream::unfold(0, |state| {
         if state <= 2 {
-            let res = ok::<_, ()>(Some((state * 2, state + 1)));
-            Either::A(delay_future(res))
+            let res: Result<_,()> = Ok((state * 2, state + 1));
+            Some(delay_future(res))
         } else {
-            Either::B(ok(None))
+            None
         }
     });
     // Creates the future with the closure
@@ -38,9 +37,9 @@ fn unfold1() {
 fn unfold_err1() {
     let mut stream = stream::unfold(0, |state| {
         if state <= 2 {
-            Ok(Some((state * 2, state + 1)))
+            Some(Ok((state * 2, state + 1)))
         } else {
-            Err(-1)
+            Some(Err(-1))
         }
     });
     sassert_next(&mut stream, 0);

tests/unsync.rs

@@ -139,7 +139,7 @@ fn mpsc_send_unpark() {
 #[test]
 fn spawn_sends_items() {
     let core = Core::new();
-    let stream = unfold(0, |i| Ok::<_,u8>(Some((i, i + 1))));
+    let stream = unfold(0, |i| Some(Ok::<_,u8>((i, i + 1))));
     let rx = mpsc::spawn(stream, &core, 1);
     assert_eq!(core.run(rx.take(4).collect()).unwrap(),
                [0, 1, 2, 3]);