Repository: rust-lang-nursery/futures-rs
Pull request #730 (2018-02-05 17:35:00)

Generalize "unfold" combinator.
unfold functions now must return a Future of an Option
rather than an Option of a future. This allows users
to hold off deciding whether or not a value should
be returned.

@alexcrichton 


Patch:
diff --git a/.travis.yml b/.travis.yml
index f3bf51c6..9c6f16b6 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -12,7 +12,7 @@ matrix:
       after_success:
         - travis-cargo doc-upload
     - os: linux
-      rust: 1.15.0
+      rust: 1.13.0
       script: cargo test
 sudo: false
 script:
diff --git a/Cargo.toml b/Cargo.toml
index c3e7760e..2e702a99 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "futures"
-version = "0.1.18"
+version = "0.1.17"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
 license = "MIT/Apache-2.0"
 readme = "README.md"
diff --git a/README.md b/README.md
index de3b99c0..e30902d0 100644
--- a/README.md
+++ b/README.md
@@ -16,7 +16,7 @@ First, add this to your `Cargo.toml`:
 
 ```toml
 [dependencies]
-futures = "0.1.17"
+futures = "0.1.14"
 ```
 
 Next, add this to your crate:
@@ -39,7 +39,7 @@ a `#[no_std]` environment, use:
 
 ```toml
 [dependencies]
-futures = { version = "0.1.17", default-features = false }
+futures = { version = "0.1.14", default-features = false }
 ```
 
 # License
diff --git a/src/future/mod.rs b/src/future/mod.rs
index 7cccd907..c858a583 100644
--- a/src/future/mod.rs
+++ b/src/future/mod.rs
@@ -967,7 +967,7 @@ pub trait Future {
     /// Create a cloneable handle to this future where all handles will resolve
     /// to the same result.
     ///
-    /// The shared() method provides a method to convert any future into a
+    /// The shared() method provides a mean to convert any future into a
     /// cloneable future. It enables a future to be polled by multiple threads.
     ///
     /// The returned `Shared` future resolves successfully with
diff --git a/src/stream/future.rs b/src/stream/future.rs
index 5b052ee4..918b2c85 100644
--- a/src/stream/future.rs
+++ b/src/stream/future.rs
@@ -14,46 +14,6 @@ pub fn new<S: Stream>(s: S) -> StreamFuture<S> {
     StreamFuture { stream: Some(s) }
 }
 
-impl<S> StreamFuture<S> {
-    /// Acquires a reference to the underlying stream that this combinator is
-    /// pulling from.
-    ///
-    /// This method returns an `Option` to account for the fact that `StreamFuture`'s
-    /// implementation of `Future::poll` consumes the underlying stream during polling 
-    /// in order to return it to the caller of `Future::poll` if the stream yielded
-    /// an element.
-    pub fn get_ref(&self) -> Option<&S> {
-        self.stream.as_ref()
-    }
-
-    /// Acquires a mutable reference to the underlying stream that this
-    /// combinator is pulling from.
-    ///
-    /// Note that care must be taken to avoid tampering with the state of the
-    /// stream which may otherwise confuse this combinator.
-    ///
-    /// This method returns an `Option` to account for the fact that `StreamFuture`'s
-    /// implementation of `Future::poll` consumes the underlying stream during polling 
-    /// in order to return it to the caller of `Future::poll` if the stream yielded
-    /// an element.
-    pub fn get_mut(&mut self) -> Option<&mut S> {
-        self.stream.as_mut()
-    }
-
-    /// Consumes this combinator, returning the underlying stream.
-    ///
-    /// Note that this may discard intermediate state of this combinator, so
-    /// care should be taken to avoid losing resources when this is called.
-    ///
-    /// This method returns an `Option` to account for the fact that `StreamFuture`'s
-    /// implementation of `Future::poll` consumes the underlying stream during polling 
-    /// in order to return it to the caller of `Future::poll` if the stream yielded
-    /// an element.
-    pub fn into_inner(self) -> Option<S> {
-        self.stream
-    }
-}
-
 impl<S: Stream> Future for StreamFuture<S> {
     type Item = (Option<S::Item>, S);
     type Error = (S::Error, S);
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index 566d143b..01ae6d69 100644
--- a/src/stream/mod.rs
+++ b/src/stream/mod.rs
@@ -115,7 +115,7 @@ if_std! {
     pub use self::chunks::Chunks;
     pub use self::collect::Collect;
     pub use self::wait::Wait;
-    pub use self::split::{SplitStream, SplitSink, ReuniteError};
+    pub use self::split::{SplitStream, SplitSink};
     pub use self::futures_unordered::FuturesUnordered;
     pub use self::futures_ordered::{futures_ordered, FuturesOrdered};
 
diff --git a/src/stream/unfold.rs b/src/stream/unfold.rs
index ac427b8c..356f1ff4 100644
--- a/src/stream/unfold.rs
+++ b/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 closure returns `None` instead of `Some(Future)`, then the `unfold()`
+/// If the future returns `None` instead of `Some`, then the `unfold()`
 /// will stop producing items and return `Ok(Async::Ready(None))` in future
 /// calls to `poll()`.
 ///
@@ -33,22 +33,23 @@ use stream::Stream;
 /// use futures::future::{self, Future};
 ///
 /// let mut stream = stream::unfold(0, |state| {
-///     if state <= 2 {
-///         let next_state = state + 1;
-///         let yielded = state  * 2;
-///         let fut = future::ok::<_, u32>((yielded, next_state));
-///         Some(fut)
-///     } else {
-///         None
-///     }
+///     future::ok::<_, u32>(
+///         if state <= 2 {
+///             let next_state = state + 1;
+///             let yielded = state  * 2;
+///             Some((yielded, next_state))
+///         } 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) -> Option<Fut>,
-          Fut: IntoFuture<Item = (It, T)>,
+    where F: FnMut(T) -> Fut,
+          Fut: IntoFuture<Item = Option<(It, T)>>,
 {
     Unfold {
         f: f,
@@ -67,8 +68,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) -> Option<Fut>,
-          Fut: IntoFuture<Item = (It, T)>,
+    where F: FnMut(T) -> Fut,
+          Fut: IntoFuture<Item = Option<(It, T)>>,
 {
     type Item = It;
     type Error = Fut::Error;
@@ -79,17 +80,18 @@ 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) => {
-                    match (self.f)(state) {
-                        Some(fut) => { self.state = State::Processing(fut.into_future()); }
-                        None => { return Ok(Async::Ready(None)); }
-                    }
+                    let fut = (self.f)(state);
+                    self.state = State::Processing(fut.into_future());
                 }
                 State::Processing(mut fut) => {
                     match fut.poll()? {
-                        Async:: Ready((item, next_state)) => {
+                        Async::Ready(Some((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);
diff --git a/src/sync/oneshot.rs b/src/sync/oneshot.rs
index d95883d3..9e79442b 100644
--- a/src/sync/oneshot.rs
+++ b/src/sync/oneshot.rs
@@ -477,7 +477,7 @@ pub fn spawn<F, E>(future: F, executor: &E) -> SpawnHandle<F::Item, F::Error>
 {
     let data = Arc::new(ExecuteInner {
         inner: Inner::new(),
-        keep_running: AtomicBool::new(false),
+        keep_running: AtomicBool::new(true),
     });
     executor.execute(Execute {
         future: future,
@@ -506,7 +506,7 @@ impl<T, E> SpawnHandle<T, E> {
     /// well if the future hasn't already resolved. This function can be used
     /// when to drop this future but keep executing the underlying future.
     pub fn forget(self) {
-        self.rx.keep_running.store(true, SeqCst);
+        self.rx.keep_running.store(false, SeqCst);
     }
 }
 
diff --git a/src/unsync/mpsc.rs b/src/unsync/mpsc.rs
index 68cf6b8c..0cff01b1 100644
--- a/src/unsync/mpsc.rs
+++ b/src/unsync/mpsc.rs
@@ -35,6 +35,7 @@ fn channel_<T>(buffer: Option<usize>) -> (Sender<T>, Receiver<T>) {
         capacity: buffer,
         blocked_senders: VecDeque::new(),
         blocked_recv: None,
+        sender_count: 1,
     }));
     let sender = Sender { shared: Rc::downgrade(&shared) };
     let receiver = Receiver { state: State::Open(shared) };
@@ -47,6 +48,8 @@ struct Shared<T> {
     capacity: Option<usize>,
     blocked_senders: VecDeque<Task>,
     blocked_recv: Option<Task>,
+    // TODO: Redundant to Rc::weak_count; use that if/when stabilized
+    sender_count: usize,
 }
 
 /// The transmission end of a channel.
@@ -61,7 +64,7 @@ impl<T> Sender<T> {
     fn do_send(&self, msg: T) -> StartSend<T, SendError<T>> {
         let shared = match self.shared.upgrade() {
             Some(shared) => shared,
-            None => return Err(SendError(msg)), // receiver was dropped
+            None => return Err(SendError(msg)),
         };
         let mut shared = shared.borrow_mut();
 
@@ -73,6 +76,7 @@ impl<T> Sender<T> {
             _ => {
                 shared.buffer.push_back(msg);
                 if let Some(task) = shared.blocked_recv.take() {
+                    drop(shared);
                     task.notify();
                 }
                 Ok(AsyncSink::Ready)
@@ -83,7 +87,11 @@ impl<T> Sender<T> {
 
 impl<T> Clone for Sender<T> {
     fn clone(&self) -> Self {
-        Sender { shared: self.shared.clone() }
+        let result = Sender { shared: self.shared.clone() };
+        if let Some(shared) = self.shared.upgrade() {
+            shared.borrow_mut().sender_count += 1;
+        }
+        result
     }
 }
 
@@ -110,9 +118,12 @@ impl<T> Drop for Sender<T> {
             Some(shared) => shared,
             None => return,
         };
-        if Rc::weak_count(&shared) == 0 {
-            if let Some(task) = shared.borrow_mut().blocked_recv.take() {
+        let mut shared = shared.borrow_mut();
+        shared.sender_count -= 1;
+        if shared.sender_count == 0 {
+            if let Some(task) = shared.blocked_recv.take() {
                 // Wake up receiver as its stream has ended
+                drop(shared);
                 task.notify();
             }
         }
diff --git a/src/unsync/oneshot.rs b/src/unsync/oneshot.rs
index 7ae2890f..c895e68a 100644
--- a/src/unsync/oneshot.rs
+++ b/src/unsync/oneshot.rs
@@ -260,7 +260,7 @@ pub fn spawn<F, E>(future: F, executor: &E) -> SpawnHandle<F::Item, F::Error>
     where F: Future,
           E: Executor<Execute<F>>,
 {
-    let flag = Rc::new(Cell::new(false));
+    let flag = Rc::new(Cell::new(true));
     let (tx, rx) = channel();
     executor.execute(Execute {
         future: future,
@@ -293,7 +293,7 @@ impl<T, E> SpawnHandle<T, E> {
     /// well if the future hasn't already resolved. This function can be used
     /// when to drop this future but keep executing the underlying future.
     pub fn forget(self) {
-        self.keep_running.set(true);
+        self.keep_running.set(false);
     }
 }
 


Test patch:
diff --git a/tests/oneshot.rs b/tests/oneshot.rs
index 45c19968..71d2d97c 100644
--- a/tests/oneshot.rs
+++ b/tests/oneshot.rs
@@ -120,134 +120,3 @@ fn cancel_sends() {
     drop(tx);
     t.join().unwrap();
 }
-
-#[test]
-fn spawn_sends_items() {
-    let core = local_executor::Core::new();
-    let future = ok::<_, ()>(1);
-    let rx = spawn(future, &core);
-    assert_eq!(core.run(rx).unwrap(), 1);
-}
-
-#[test]
-fn spawn_kill_dead_stream() {
-    use std::thread;
-    use std::time::Duration;
-    use futures::future::Either;
-    use futures::sync::oneshot;
-
-    // a future which never returns anything (forever accepting incoming
-    // connections), but dropping it leads to observable side effects
-    // (like closing listening sockets, releasing limited resources,
-    // ...)
-    #[derive(Debug)]
-    struct Dead {
-        // when dropped you should get Err(oneshot::Canceled) on the
-        // receiving end
-        done: oneshot::Sender<()>,
-    }
-    impl Future for Dead {
-        type Item = ();
-        type Error = ();
-
-        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-            Ok(Async::NotReady)
-        }
-    }
-
-    // need to implement a timeout for the test, as it would hang
-    // forever right now
-    let (timeout_tx, timeout_rx) = oneshot::channel();
-    thread::spawn(move || {
-        thread::sleep(Duration::from_millis(1000));
-        let _ = timeout_tx.send(());
-    });
-
-    let core = local_executor::Core::new();
-    let (done_tx, done_rx) = oneshot::channel();
-    let future = Dead{done: done_tx};
-    let rx = spawn(future, &core);
-    let res = core.run(
-        Ok::<_, ()>(())
-        .into_future()
-        .then(move |_| {
-            // now drop the spawned future: maybe some timeout exceeded,
-            // or some connection on this end was closed by the remote
-            // end.
-            drop(rx);
-            // and wait for the spawned future to release its resources
-            done_rx
-        })
-        .select2(timeout_rx)
-    );
-    match res {
-        Err(Either::A((oneshot::Canceled, _))) => (),
-        Ok(Either::B(((), _))) => {
-            panic!("dead future wasn't canceled (timeout)");
-        },
-        _ => {
-            panic!("dead future wasn't canceled (unexpected result)");
-        },
-    }
-}
-
-#[test]
-fn spawn_dont_kill_forgot_dead_stream() {
-    use std::thread;
-    use std::time::Duration;
-    use futures::future::Either;
-    use futures::sync::oneshot;
-
-    // a future which never returns anything (forever accepting incoming
-    // connections), but dropping it leads to observable side effects
-    // (like closing listening sockets, releasing limited resources,
-    // ...)
-    #[derive(Debug)]
-    struct Dead {
-        // when dropped you should get Err(oneshot::Canceled) on the
-        // receiving end
-        done: oneshot::Sender<()>,
-    }
-    impl Future for Dead {
-        type Item = ();
-        type Error = ();
-
-        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-            Ok(Async::NotReady)
-        }
-    }
-
-    // need to implement a timeout for the test, as it would hang
-    // forever right now
-    let (timeout_tx, timeout_rx) = oneshot::channel();
-    thread::spawn(move || {
-        thread::sleep(Duration::from_millis(1000));
-        let _ = timeout_tx.send(());
-    });
-
-    let core = local_executor::Core::new();
-    let (done_tx, done_rx) = oneshot::channel();
-    let future = Dead{done: done_tx};
-    let rx = spawn(future, &core);
-    let res = core.run(
-        Ok::<_, ()>(())
-        .into_future()
-        .then(move |_| {
-            // forget the spawned future: should keep running, i.e. hit
-            // the timeout below.
-            rx.forget();
-            // and wait for the spawned future to release its resources
-            done_rx
-        })
-        .select2(timeout_rx)
-    );
-    match res {
-        Err(Either::A((oneshot::Canceled, _))) => {
-            panic!("forgotten dead future was canceled");
-        },
-        Ok(Either::B(((), _))) => (), // reached timeout
-        _ => {
-            panic!("forgotten dead future was canceled (unexpected result)");
-        },
-    }
-}
diff --git a/tests/unfold.rs b/tests/unfold.rs
index 1669a18a..eef27160 100644
--- a/tests/unfold.rs
+++ b/tests/unfold.rs
@@ -3,6 +3,7 @@ extern crate futures;
 mod support;
 
 use futures::stream;
+use futures::future::{ok, Either};
 
 use support::*;
 
@@ -10,10 +11,10 @@ use support::*;
 fn unfold1() {
     let mut stream = stream::unfold(0, |state| {
         if state <= 2 {
-            let res: Result<_,()> = Ok((state * 2, state + 1));
-            Some(delay_future(res))
+            let res = ok::<_, ()>(Some((state * 2, state + 1)));
+            Either::A(delay_future(res))
         } else {
-            None
+            Either::B(ok(None))
         }
     });
     // Creates the future with the closure
@@ -37,9 +38,9 @@ fn unfold1() {
 fn unfold_err1() {
     let mut stream = stream::unfold(0, |state| {
         if state <= 2 {
-            Some(Ok((state * 2, state + 1)))
+            Ok(Some((state * 2, state + 1)))
         } else {
-            Some(Err(-1))
+            Err(-1)
         }
     });
     sassert_next(&mut stream, 0);
diff --git a/tests/unsync-oneshot.rs b/tests/unsync-oneshot.rs
index 55b0ca5a..7f19d830 100644
--- a/tests/unsync-oneshot.rs
+++ b/tests/unsync-oneshot.rs
@@ -2,10 +2,7 @@ extern crate futures;
 
 use futures::prelude::*;
 use futures::future;
-use futures::unsync::oneshot::{channel, Canceled, spawn};
-
-mod support;
-use support::local_executor;
+use futures::unsync::oneshot::{channel, Canceled};
 
 #[test]
 fn smoke() {
@@ -56,134 +53,3 @@ fn is_canceled() {
     drop(rx);
     assert!(tx.is_canceled());
 }
-
-#[test]
-fn spawn_sends_items() {
-    let core = local_executor::Core::new();
-    let future = future::ok::<_, ()>(1);
-    let rx = spawn(future, &core);
-    assert_eq!(core.run(rx).unwrap(), 1);
-}
-
-#[test]
-fn spawn_kill_dead_stream() {
-    use std::thread;
-    use std::time::Duration;
-    use futures::future::Either;
-    use futures::sync::oneshot;
-
-    // a future which never returns anything (forever accepting incoming
-    // connections), but dropping it leads to observable side effects
-    // (like closing listening sockets, releasing limited resources,
-    // ...)
-    #[derive(Debug)]
-    struct Dead {
-        // when dropped you should get Err(oneshot::Canceled) on the
-        // receiving end
-        done: oneshot::Sender<()>,
-    }
-    impl Future for Dead {
-        type Item = ();
-        type Error = ();
-
-        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-            Ok(Async::NotReady)
-        }
-    }
-
-    // need to implement a timeout for the test, as it would hang
-    // forever right now
-    let (timeout_tx, timeout_rx) = oneshot::channel();
-    thread::spawn(move || {
-        thread::sleep(Duration::from_millis(1000));
-        let _ = timeout_tx.send(());
-    });
-
-    let core = local_executor::Core::new();
-    let (done_tx, done_rx) = oneshot::channel();
-    let future = Dead{done: done_tx};
-    let rx = spawn(future, &core);
-    let res = core.run(
-        Ok::<_, ()>(())
-        .into_future()
-        .then(move |_| {
-            // now drop the spawned future: maybe some timeout exceeded,
-            // or some connection on this end was closed by the remote
-            // end.
-            drop(rx);
-            // and wait for the spawned future to release its resources
-            done_rx
-        })
-        .select2(timeout_rx)
-    );
-    match res {
-        Err(Either::A((oneshot::Canceled, _))) => (),
-        Ok(Either::B(((), _))) => {
-            panic!("dead future wasn't canceled (timeout)");
-        },
-        _ => {
-            panic!("dead future wasn't canceled (unexpected result)");
-        },
-    }
-}
-
-#[test]
-fn spawn_dont_kill_forgot_dead_stream() {
-    use std::thread;
-    use std::time::Duration;
-    use futures::future::Either;
-    use futures::sync::oneshot;
-
-    // a future which never returns anything (forever accepting incoming
-    // connections), but dropping it leads to observable side effects
-    // (like closing listening sockets, releasing limited resources,
-    // ...)
-    #[derive(Debug)]
-    struct Dead {
-        // when dropped you should get Err(oneshot::Canceled) on the
-        // receiving end
-        done: oneshot::Sender<()>,
-    }
-    impl Future for Dead {
-        type Item = ();
-        type Error = ();
-
-        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-            Ok(Async::NotReady)
-        }
-    }
-
-    // need to implement a timeout for the test, as it would hang
-    // forever right now
-    let (timeout_tx, timeout_rx) = oneshot::channel();
-    thread::spawn(move || {
-        thread::sleep(Duration::from_millis(1000));
-        let _ = timeout_tx.send(());
-    });
-
-    let core = local_executor::Core::new();
-    let (done_tx, done_rx) = oneshot::channel();
-    let future = Dead{done: done_tx};
-    let rx = spawn(future, &core);
-    let res = core.run(
-        Ok::<_, ()>(())
-        .into_future()
-        .then(move |_| {
-            // forget the spawned future: should keep running, i.e. hit
-            // the timeout below.
-            rx.forget();
-            // and wait for the spawned future to release its resources
-            done_rx
-        })
-        .select2(timeout_rx)
-    );
-    match res {
-        Err(Either::A((oneshot::Canceled, _))) => {
-            panic!("forgotten dead future was canceled");
-        },
-        Ok(Either::B(((), _))) => (), // reached timeout
-        _ => {
-            panic!("forgotten dead future was canceled (unexpected result)");
-        },
-    }
-}
diff --git a/tests/unsync.rs b/tests/unsync.rs
index b5ae8d0f..5af933c1 100644
--- a/tests/unsync.rs
+++ b/tests/unsync.rs
@@ -43,29 +43,6 @@ fn mpsc_rx_end() {
     }).wait().unwrap();
 }
 
-#[test]
-fn mpsc_tx_clone_weak_rc() {
-    let (tx, mut rx) = mpsc::channel::<i32>(1); // rc = 1
-
-    let tx_clone = tx.clone(); // rc = 2
-    lazy(|| {
-        assert_eq!(rx.poll().unwrap(), Async::NotReady);
-        Ok(()) as Result<(), ()>
-    }).wait().unwrap();
-
-    drop(tx); // rc = 1
-    lazy(|| {
-        assert_eq!(rx.poll().unwrap(), Async::NotReady);
-        Ok(()) as Result<(), ()>
-    }).wait().unwrap();
-
-    drop(tx_clone); // rc = 0
-    lazy(|| {
-        assert_eq!(rx.poll().unwrap(), Async::Ready(None));
-        Ok(()) as Result<(), ()>
-    }).wait().unwrap();
-}
-
 #[test]
 fn mpsc_tx_notready() {
     let (tx, _rx) = mpsc::channel::<i32>(1);
@@ -139,7 +116,7 @@ fn mpsc_send_unpark() {
 #[test]
 fn spawn_sends_items() {
     let core = Core::new();
-    let stream = unfold(0, |i| Some(Ok::<_,u8>((i, i + 1))));
+    let stream = unfold(0, |i| Ok::<_,u8>(Some((i, i + 1))));
     let rx = mpsc::spawn(stream, &core, 1);
     assert_eq!(core.run(rx.take(4).collect()).unwrap(),
                [0, 1, 2, 3]);