Repository: rust-lang-nursery/futures-rs
Pull request #727 (2018-02-05 17:33:08)

Return Option in Fluch accessors instead of panic
@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/sink/flush.rs b/src/sink/flush.rs
index f66811e0..ed152931 100644
--- a/src/sink/flush.rs
+++ b/src/sink/flush.rs
@@ -15,18 +15,21 @@ pub fn new<S: Sink>(sink: S) -> Flush<S> {
 
 impl<S: Sink> Flush<S> {
     /// Get a shared reference to the inner sink.
-    pub fn get_ref(&self) -> &S {
-        self.sink.as_ref().expect("Attempted `Flush::get_ref` after the flush completed")
+    /// Returns `None` if the sink has already been flushed.
+    pub fn get_ref(&self) -> Option<&S> {
+        self.sink.as_ref()
     }
 
     /// Get a mutable reference to the inner sink.
-    pub fn get_mut(&mut self) -> &mut S {
-        self.sink.as_mut().expect("Attempted `Flush::get_mut` after the flush completed")
+    /// Returns `None` if the sink has already been flushed.
+    pub fn get_mut(&mut self) -> Option<&mut S> {
+        self.sink.as_mut()
     }
 
     /// Consume the `Flush` and return the inner sink.
-    pub fn into_inner(self) -> S {
-        self.sink.expect("Attempted `Flush::into_inner` after the flush completed")
+    /// Returns `None` if the sink has already been flushed.
+    pub fn into_inner(self) -> Option<S> {
+        self.sink
     }
 }
 
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/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/sink.rs b/tests/sink.rs
index cb2fdcf2..4d5355b1 100644
--- a/tests/sink.rs
+++ b/tests/sink.rs
@@ -242,7 +242,7 @@ fn with_flush_propagate() {
     let mut task = executor::spawn(sink.flush());
     assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
     assert!(!flag.get());
-    assert_eq!(task.get_mut().get_mut().get_mut().force_flush(), vec![0, 1]);
+    assert_eq!(task.get_mut().get_mut().unwrap().get_mut().force_flush(), vec![0, 1]);
     assert!(flag.get());
     assert!(task.poll_future_notify(&flag, 0).unwrap().is_ready());
 }
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..e86b95ca 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);