Merge pull request #1240 from carllerche/fix-mpsc-poll-complete

fix mpsc::Sender::poll_complete impl

Author: MajorBreakfast

benches/sync_mpsc.rs

@@ -1,5 +1,6 @@
 #![feature(test)]
 
+#[macro_use]
 extern crate futures;
 extern crate test;
 
@@ -106,7 +107,6 @@ impl Stream for TestSender {
             Err(_) => panic!(),
             Ok(AsyncSink::Ready) => {
                 self.last += 1;
-                assert_eq!(Ok(Async::Ready(())), self.tx.poll_complete());
                 Ok(Async::Ready(Some(self.last)))
             }
             Ok(AsyncSink::NotReady(_)) => {

src/sync/mpsc/mod.rs

@@ -649,7 +649,14 @@ impl<T> Sink for Sender<T> {
     }
 
     fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
-        Ok(Async::Ready(()))
+        self.poll_ready()
+            // At this point, the value cannot be returned and `SendError`
+            // cannot be created with a `T` without breaking backwards
+            // comptibility. This means we cannot return an error.
+            //
+            // That said, there is also no guarantee that a `poll_complete`
+            // returning `Ok` implies the receiver sees the message.
+            .or_else(|_| Ok(().into()))
     }
 
     fn close(&mut self) -> Poll<(), SendError<T>> {

tests/mpsc.rs

@@ -548,3 +548,19 @@ fn try_send_fail() {
     assert_eq!(rx.next(), Some(Ok("goodbye")));
     assert!(rx.next().is_none());
 }
+
+#[test]
+fn bounded_is_really_bounded() {
+    use futures::Async::*;
+    let (mut tx, mut rx) = mpsc::channel(0);
+    lazy(|| {
+        assert!(tx.start_send(1).unwrap().is_ready());
+        // Not ready until we receive
+        assert!(!tx.poll_complete().unwrap().is_ready());
+        // Receive the value
+        assert_eq!(rx.poll().unwrap(), Ready(Some(1)));
+        // Now the sender is ready
+        assert!(tx.poll_complete().unwrap().is_ready());
+        Ok::<_, ()>(())
+    }).wait().unwrap();
+}

tests/sink.rs

@@ -392,6 +392,12 @@ fn fanout_backpressure() {
     let (item, right_recv) = right_recv.into_future().wait().unwrap();
     assert_eq!(item, Some(1));
     assert!(flag.get());
+    let (item, left_recv) = left_recv.into_future().wait().unwrap();
+    assert_eq!(item, Some(2));
+    assert!(flag.get());
+    assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
+    let (item, right_recv) = right_recv.into_future().wait().unwrap();
+    assert_eq!(item, Some(2));
     match task.poll_future_notify(&flag, 0).unwrap() {
         Async::Ready(_) => {
         },