Repository: alexcrichton/futures-rs
Pull request #366 (2017-01-19 18:42:36)

Rename oneshot::Sender::complete to `send`
This commit renames the `complete` method to `send`. At the same time it also
updates the return value to `Result<(), T>` to express situations where the item
failed to get sent. A `#[deprecated]` marker is left on the old `complete`
method so existing code continues to compile, however.

Closes https://github.com/alexcrichton/futures-rs/issues/317


Patch:
diff --git a/futures-cpupool/src/lib.rs b/futures-cpupool/src/lib.rs
index d4a803d8..946b3770 100644
--- a/futures-cpupool/src/lib.rs
+++ b/futures-cpupool/src/lib.rs
@@ -283,7 +283,10 @@ impl<F: Future> Future for MySender<F, Result<F::Item, F::Error>> {
             Ok(Async::NotReady) => return Ok(Async::NotReady),
             Err(e) => Err(e),
         };
-        self.tx.take().unwrap().complete(res);
+
+        // if the receiving end has gone away then that's ok, we just ignore the
+        // send error here.
+        drop(self.tx.take().unwrap().send(res));
         Ok(Async::Ready(()))
     }
 }
diff --git a/src/sync/oneshot.rs b/src/sync/oneshot.rs
index d744ee36..2b44aa59 100644
--- a/src/sync/oneshot.rs
+++ b/src/sync/oneshot.rs
@@ -91,7 +91,7 @@ struct Inner<T> {
 ///     }).wait();
 /// });
 ///
-/// c.complete(3);
+/// c.send(3).unwrap();
 /// ```
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     let inner = Arc::new(Inner {
@@ -110,13 +110,28 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
 }
 
 impl<T> Sender<T> {
+    #[deprecated(note = "renamed to `send`", since = "0.1.11")]
+    #[doc(hidden)]
+    #[cfg(feature = "with-deprecated")]
+    pub fn complete(self, t: T) {
+        drop(self.send(t));
+    }
+
     /// Completes this oneshot with a successful result.
     ///
     /// This function will consume `self` and indicate to the other end, the
     /// `Receiver`, that the error provided is the result of the computation this
     /// represents.
-    pub fn complete(self, t: T) {
-        // First up, flag that this method was called and then store the data.
+    ///
+    /// If the value is successfully enqueued for the remote end to receive,
+    /// then `Ok(())` is returned. If the receiving end was deallocated before
+    /// this function was called, however, then `Err` is returned with the value
+    /// provided.
+    pub fn send(self, t: T) -> Result<(), T> {
+        if self.inner.complete.load(SeqCst) {
+            return Err(t)
+        }
+
         // Note that this lock acquisition should always succeed as it can only
         // interfere with `poll` in `Receiver` which is only called when the
         // `complete` flag is true, which we're setting here.
@@ -124,6 +139,7 @@ impl<T> Sender<T> {
         assert!(slot.is_none());
         *slot = Some(t);
         drop(slot);
+        Ok(())
     }
 
     /// Polls this `Sender` half to detect whether the `Receiver` this has
diff --git a/src/unsync/oneshot.rs b/src/unsync/oneshot.rs
index 555a6b6e..6008efa6 100644
--- a/src/unsync/oneshot.rs
+++ b/src/unsync/oneshot.rs
@@ -74,9 +74,17 @@ impl<T> Sender<T> {
     /// This function will consume `self` and indicate to the other end, the
     /// `Receiver`, that the error provided is the result of the computation this
     /// represents.
-    pub fn complete(self, val: T) {
+    ///
+    /// If the value is successfully enqueued for the remote end to receive,
+    /// then `Ok(())` is returned. If the receiving end was deallocated before
+    /// this function was called, however, then `Err` is returned with the value
+    /// provided.
+    pub fn send(self, val: T) -> Result<(), T> {
         if let Some(inner) = self.inner.upgrade() {
             inner.borrow_mut().value = Some(val);
+            Ok(())
+        } else {
+            Err(val)
         }
     }
 


Test patch:
diff --git a/tests/all.rs b/tests/all.rs
index 3bf201d8..bc872bb3 100644
--- a/tests/all.rs
+++ b/tests/all.rs
@@ -110,7 +110,7 @@ fn flatten() {
 fn smoke_oneshot() {
     assert_done(|| {
         let (c, p) = oneshot::channel();
-        c.complete(1);
+        c.send(1).unwrap();
         p
     }, Ok(1));
     assert_done(|| {
@@ -149,7 +149,7 @@ fn select_cancels() {
     // assert!(f.poll(&mut Task::new()).is_not_ready());
     assert!(brx.try_recv().is_err());
     assert!(drx.try_recv().is_err());
-    a.complete(1);
+    a.send(1).unwrap();
     let res = executor::spawn(f).poll_future(unpark_panic());
     assert!(res.ok().unwrap().is_ready());
     assert_eq!(brx.recv().unwrap(), 1);
@@ -164,7 +164,7 @@ fn select_cancels() {
     let mut f = executor::spawn(b.select(d).then(unselect));
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
-    a.complete(1);
+    a.send(1).unwrap();
     assert!(f.poll_future(unpark_panic()).ok().unwrap().is_ready());
     drop((c, f));
     assert!(drx.recv().is_err());
@@ -210,7 +210,7 @@ fn join_incomplete() {
     let mut f = executor::spawn(ok(1).join(b).map(move |r| tx.send(r).unwrap()));
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
     assert!(rx.try_recv().is_err());
-    a.complete(2);
+    a.send(2).unwrap();
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_ready());
     assert_eq!(rx.recv().unwrap(), (1, 2));
 
@@ -219,7 +219,7 @@ fn join_incomplete() {
     let mut f = executor::spawn(b.join(Ok(2)).map(move |r| tx.send(r).unwrap()));
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
     assert!(rx.try_recv().is_err());
-    a.complete(1);
+    a.send(1).unwrap();
     assert!(f.poll_future(unpark_noop()).ok().unwrap().is_ready());
     assert_eq!(rx.recv().unwrap(), (1, 2));
 
@@ -278,7 +278,7 @@ fn select2() {
         let f = b.select(d);
         let (tx, rx) = channel();
         f.map(move |r| tx.send(r).unwrap()).forget();
-        a.complete(1);
+        a.send(1).unwrap();
         let (val, next) = rx.recv().unwrap();
         assert_eq!(val, 1);
         let (tx, rx) = channel();
@@ -300,7 +300,7 @@ fn select2() {
         let (tx, rx) = channel();
         next.map(move |r| tx.send(r).unwrap()).forget();
         assert_eq!(rx.try_recv().err().unwrap(), TryRecvError::Empty);
-        a.complete(2);
+        a.send(2).unwrap();
         assert_eq!(rx.recv().unwrap(), 2);
     }
 
diff --git a/tests/buffer_unordered.rs b/tests/buffer_unordered.rs
index 8874c291..e47a0d8d 100644
--- a/tests/buffer_unordered.rs
+++ b/tests/buffer_unordered.rs
@@ -42,14 +42,14 @@ fn works() {
     let o4 = rx3.recv().unwrap();
     assert!(rx4.try_recv().is_err());
 
-    o1.complete(1);
+    o1.send(1).unwrap();
     assert_eq!(rx4.recv(), Ok(1));
-    o3.complete(3);
+    o3.send(3).unwrap();
     assert_eq!(rx4.recv(), Ok(3));
     tx2.send(()).unwrap();
-    o2.complete(2);
+    o2.send(2).unwrap();
     assert_eq!(rx4.recv(), Ok(2));
-    o4.complete(4);
+    o4.send(4).unwrap();
     assert_eq!(rx4.recv(), Ok(4));
 
     let o5 = rx3.recv().unwrap();
@@ -58,15 +58,15 @@ fn works() {
     let o8 = rx3.recv().unwrap();
     let o9 = rx3.recv().unwrap();
 
-    o5.complete(5);
+    o5.send(5).unwrap();
     assert_eq!(rx4.recv(), Ok(5));
-    o8.complete(8);
+    o8.send(8).unwrap();
     assert_eq!(rx4.recv(), Ok(8));
-    o9.complete(9);
+    o9.send(9).unwrap();
     assert_eq!(rx4.recv(), Ok(9));
-    o7.complete(7);
+    o7.send(7).unwrap();
     assert_eq!(rx4.recv(), Ok(7));
-    o6.complete(6);
+    o6.send(6).unwrap();
     assert_eq!(rx4.recv(), Ok(6));
 
     t1.join().unwrap();
diff --git a/tests/eager_drop.rs b/tests/eager_drop.rs
index 507a9fdd..94f30d2b 100644
--- a/tests/eager_drop.rs
+++ b/tests/eager_drop.rs
@@ -60,7 +60,7 @@ fn and_then_drops_eagerly() {
         ok(1)
     }).forget();
     assert!(rx2.try_recv().is_err());
-    c.complete(());
+    c.send(()).unwrap();
     rx2.recv().unwrap();
 }
 
diff --git a/tests/eventual.rs b/tests/eventual.rs
index feb4606b..5bf3680d 100644
--- a/tests/eventual.rs
+++ b/tests/eventual.rs
@@ -47,7 +47,7 @@ fn and_then2() {
 #[test]
 fn oneshot1() {
     let (c, p) = oneshot::channel::<i32>();
-    let t = thread::spawn(|| c.complete(1));
+    let t = thread::spawn(|| c.send(1).unwrap());
 
     let (tx, rx) = mpsc::channel();
     p.map(move |e| tx.send(e).unwrap()).forget();
@@ -58,7 +58,7 @@ fn oneshot1() {
 #[test]
 fn oneshot2() {
     let (c, p) = oneshot::channel::<i32>();
-    let t = thread::spawn(|| c.complete(1));
+    let t = thread::spawn(|| c.send(1).unwrap());
     t.join().unwrap();
 
     let (tx, rx) = mpsc::channel();
@@ -72,7 +72,7 @@ fn oneshot3() {
     let (tx, rx) = mpsc::channel();
     p.map(move |e| tx.send(e).unwrap()).forget();
 
-    let t = thread::spawn(|| c.complete(1));
+    let t = thread::spawn(|| c.send(1).unwrap());
     t.join().unwrap();
 
     assert_eq!(rx.recv(), Ok(1));
@@ -102,7 +102,7 @@ fn oneshot5() {
 fn oneshot6() {
     let (c, p) = oneshot::channel::<i32>();
     drop(p);
-    c.complete(2);
+    c.send(2).unwrap_err();
 }
 
 #[test]
@@ -145,7 +145,7 @@ fn or_else1() {
 
     assert!(rx.try_recv().is_err());
     drop(c1);
-    c2.complete(3);
+    c2.send(3).unwrap();
     p1.or_else(|_| p2).map(move |v| tx.send(v).unwrap()).forget();
 
     assert_eq!(rx.recv(), Ok(2));
@@ -165,7 +165,7 @@ fn or_else2() {
         ok::<i32, i32>(1)
     }).forget();
 
-    c1.complete(2);
+    c1.send(2).unwrap();
     assert!(rx.recv().is_err());
 }
 
@@ -186,9 +186,9 @@ fn join2() {
     let (tx, rx) = mpsc::channel();
     p1.join(p2).map(move |v| tx.send(v).unwrap()).forget();
     assert!(rx.try_recv().is_err());
-    c1.complete(1);
+    c1.send(1).unwrap();
     assert!(rx.try_recv().is_err());
-    c2.complete(2);
+    c2.send(2).unwrap();
     assert_eq!(rx.recv(), Ok((1, 2)));
     assert!(rx.recv().is_err());
 }
@@ -227,11 +227,11 @@ fn join5() {
     let (tx, rx) = mpsc::channel();
     p1.join(p2).join(p3).map(move |v| tx.send(v).unwrap()).forget();
     assert!(rx.try_recv().is_err());
-    c1.complete(1);
+    c1.send(1).unwrap();
     assert!(rx.try_recv().is_err());
-    c2.complete(2);
+    c2.send(2).unwrap();
     assert!(rx.try_recv().is_err());
-    c3.complete(3);
+    c3.send(3).unwrap();
     assert_eq!(rx.recv(), Ok(((1, 2), 3)));
     assert!(rx.recv().is_err());
 }
@@ -243,14 +243,14 @@ fn select1() {
     let (tx, rx) = mpsc::channel();
     p1.select(p2).map(move |v| tx.send(v).unwrap()).forget();
     assert!(rx.try_recv().is_err());
-    c1.complete(1);
+    c1.send(1).unwrap();
     let (v, p2) = rx.recv().unwrap();
     assert_eq!(v, 1);
     assert!(rx.recv().is_err());
 
     let (tx, rx) = mpsc::channel();
     p2.map(move |v| tx.send(v).unwrap()).forget();
-    c2.complete(2);
+    c2.send(2).unwrap();
     assert_eq!(rx.recv(), Ok(2));
     assert!(rx.recv().is_err());
 }
@@ -269,7 +269,7 @@ fn select2() {
 
     let (tx, rx) = mpsc::channel();
     p2.map(move |v| tx.send(v).unwrap()).forget();
-    c2.complete(2);
+    c2.send(2).unwrap();
     assert_eq!(rx.recv(), Ok(2));
     assert!(rx.recv().is_err());
 }
@@ -299,7 +299,7 @@ fn select4() {
 
     let t = thread::spawn(move || {
         for c in rx {
-            c.complete(1);
+            c.send(1).unwrap();
         }
     });
 
diff --git a/tests/futures_unordered.rs b/tests/futures_unordered.rs
index 8b9b29f8..d877f03b 100644
--- a/tests/futures_unordered.rs
+++ b/tests/futures_unordered.rs
@@ -17,11 +17,11 @@ fn works_1() {
     let stream = futures_unordered(vec![a_rx, b_rx, c_rx]);
 
     let mut spawn = futures::executor::spawn(stream);
-    b_tx.complete(99);
+    b_tx.send(99).unwrap();
     assert_eq!(Some(Ok(99)), spawn.wait_stream());
 
-    a_tx.complete(33);
-    c_tx.complete(33);
+    a_tx.send(33).unwrap();
+    c_tx.send(33).unwrap();
     assert_eq!(Some(Ok(33)), spawn.wait_stream());
     assert_eq!(Some(Ok(33)), spawn.wait_stream());
     assert_eq!(None, spawn.wait_stream());
@@ -36,10 +36,10 @@ fn works_2() {
     let stream = futures_unordered(vec![a_rx.boxed(), b_rx.join(c_rx).map(|(a, b)| a + b).boxed()]);
 
     let mut spawn = futures::executor::spawn(stream);
-    a_tx.complete(33);
-    b_tx.complete(33);
+    a_tx.send(33).unwrap();
+    b_tx.send(33).unwrap();
     assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_ready());
-    c_tx.complete(33);
+    c_tx.send(33).unwrap();
     assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_ready());
 }
 
@@ -59,10 +59,10 @@ fn finished_future_ok() {
         assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_not_ready());
     }
 
-    b_tx.complete(Box::new(()));
+    b_tx.send(Box::new(())).unwrap();
     let next = spawn.poll_stream(support::unpark_noop()).unwrap();
     assert!(next.is_ready());
-    c_tx.complete(Box::new(()));
+    c_tx.send(Box::new(())).unwrap();
     assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_not_ready());
     assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_not_ready());
 }
diff --git a/tests/shared.rs b/tests/shared.rs
index bacb9ad5..f74f5a94 100644
--- a/tests/shared.rs
+++ b/tests/shared.rs
@@ -19,7 +19,7 @@ fn send_shared_oneshot_and_wait_on_multiple_threads(threads_number: u32) {
             assert!(*cloned_future.wait().unwrap() == 6);
         })
     }).collect::<Vec<_>>();
-    tx.complete(6);
+    tx.send(6).unwrap();
     assert!(*f.wait().unwrap() == 6);
     for f in threads {
         f.join().unwrap();
@@ -57,13 +57,13 @@ fn drop_on_one_task_ok() {
     let (tx3, rx3) = oneshot::channel::<u32>();
 
     let t2 = thread::spawn(|| {
-        drop(f2.map(|x| tx3.complete(*x)).map_err(|_| ()).wait());
+        drop(f2.map(|x| tx3.send(*x).unwrap()).map_err(|_| ()).wait());
     });
 
-    tx2.complete(11); // cancel `f1`
+    tx2.send(11).unwrap(); // cancel `f1`
     t1.join().unwrap();
 
-    tx.complete(42); // Should cause `f2` and then `rx3` to get resolved.
+    tx.send(42).unwrap(); // Should cause `f2` and then `rx3` to get resolved.
     let result = rx3.wait().unwrap();
     assert_eq!(result, 42);
     t2.join().unwrap();
@@ -97,7 +97,7 @@ fn peek() {
     }
 
     // Completing the underlying future has no effect, because the value has not been `poll`ed in.
-    tx0.complete(42);
+    tx0.send(42).unwrap();
     for _ in 0..2 {
         assert!(f1.peek().is_none());
         assert!(f2.peek().is_none());
@@ -123,20 +123,20 @@ fn polled_then_ignored() {
     let (tx2, rx2) = oneshot::channel::<u32>();
     let (tx3, rx3) = oneshot::channel::<u32>();
 
-    core.spawn(f1.map(|n| tx3.complete(*n)).map_err(|_|()));
+    core.spawn(f1.map(|n| tx3.send(*n).unwrap()).map_err(|_|()));
 
     core.run(future::ok::<(),()>(())).unwrap(); // Allow f1 to be polled.
 
     core.spawn(f2.map_err(|_| ()).map(|x| *x).select(rx2.map_err(|_| ())).map_err(|_| ())
-        .and_then(|(_, f2)| rx3.map_err(|_| ()).map(move |n| {drop(f2); tx1.complete(n)})));
+        .and_then(|(_, f2)| rx3.map_err(|_| ()).map(move |n| {drop(f2); tx1.send(n).unwrap()})));
 
     core.run(future::ok::<(),()>(())).unwrap();  // Allow f2 to be polled.
 
-    tx2.complete(11); // Resolve rx2, causing f2 to no longer get polled.
+    tx2.send(11).unwrap(); // Resolve rx2, causing f2 to no longer get polled.
 
-    core.run(future::ok::<(),()>(())).unwrap(); // Let the complete() propagate.
+    core.run(future::ok::<(),()>(())).unwrap(); // Let the send() propagate.
 
-    tx0.complete(42); // Should cause f1, then rx3, and then rx1 to resolve.
+    tx0.send(42).unwrap(); // Should cause f1, then rx3, and then rx1 to resolve.
 
     assert_eq!(core.run(rx1).unwrap(), 42);
 }
@@ -166,7 +166,7 @@ fn recursive_poll() {
     // deadlock or panic due to a recursive lock() on a mutex.
     core.run(future::ok::<(),()>(())).unwrap();
 
-    tx1.complete(()); // Break the cycle.
+    tx1.send(()).unwrap(); // Break the cycle.
     drop(tx0);
     core.run(f3).unwrap();
 }
@@ -198,7 +198,7 @@ fn recursive_poll_with_unpark() {
     // deadlock or panic due to a recursive lock() on a mutex.
     core.run(future::ok::<(),()>(())).unwrap();
 
-    tx1.complete(()); // Break the cycle.
+    tx1.send(()).unwrap(); // Break the cycle.
     drop(tx0);
     core.run(f3).unwrap();
 }
diff --git a/tests/sink.rs b/tests/sink.rs
index ebd0c431..97283a79 100644
--- a/tests/sink.rs
+++ b/tests/sink.rs
@@ -144,7 +144,7 @@ fn with_flush() {
     let flag = Flag::new();
     let mut task = executor::spawn(sink.flush());
     assert!(task.poll_future(flag.clone()).unwrap().is_not_ready());
-    tx.complete(());
+    tx.send(()).unwrap();
     assert!(flag.get());
 
     let sink = match task.poll_future(flag.clone()).unwrap() {
diff --git a/tests/stream.rs b/tests/stream.rs
index 9f9971bc..71ecb866 100644
--- a/tests/stream.rs
+++ b/tests/stream.rs
@@ -165,9 +165,9 @@ fn buffered() {
 
     let mut rx = rx.buffered(2);
     sassert_empty(&mut rx);
-    c.complete(3);
+    c.send(3).unwrap();
     sassert_empty(&mut rx);
-    a.complete(5);
+    a.send(5).unwrap();
     let mut rx = rx.wait();
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), Some(Ok(3)));
@@ -183,9 +183,9 @@ fn buffered() {
 
     let mut rx = rx.buffered(1);
     sassert_empty(&mut rx);
-    c.complete(3);
+    c.send(3).unwrap();
     sassert_empty(&mut rx);
-    a.complete(5);
+    a.send(5).unwrap();
     let mut rx = rx.wait();
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), Some(Ok(3)));
@@ -205,9 +205,9 @@ fn unordered() {
     let mut rx = rx.buffer_unordered(2);
     sassert_empty(&mut rx);
     let mut rx = rx.wait();
-    c.complete(3);
+    c.send(3).unwrap();
     assert_eq!(rx.next(), Some(Ok(3)));
-    a.complete(5);
+    a.send(5).unwrap();
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), None);
 
@@ -222,9 +222,9 @@ fn unordered() {
     // We don't even get to see `c` until `a` completes.
     let mut rx = rx.buffer_unordered(1);
     sassert_empty(&mut rx);
-    c.complete(3);
+    c.send(3).unwrap();
     sassert_empty(&mut rx);
-    a.complete(5);
+    a.send(5).unwrap();
     let mut rx = rx.wait();
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), Some(Ok(3)));
diff --git a/tests/unsync-oneshot.rs b/tests/unsync-oneshot.rs
index 38c6d97e..565c3953 100644
--- a/tests/unsync-oneshot.rs
+++ b/tests/unsync-oneshot.rs
@@ -7,7 +7,7 @@ use futures::unsync::oneshot::{channel, Canceled};
 #[test]
 fn smoke() {
     let (tx, rx) = channel();
-    tx.complete(33);
+    tx.send(33).unwrap();
     assert_eq!(rx.wait().unwrap(), 33);
 }
 
@@ -28,7 +28,7 @@ fn tx_complete_rx_unparked() {
     let (tx, rx) = channel();
 
     let res = rx.join(future::lazy(move || {
-        tx.complete(55);
+        tx.send(55).unwrap();
         Ok(11)
     }));
     assert_eq!(res.wait().unwrap(), (55, 11));