Add Drop impl to futures::channel::mpsc::UnboundedReceiver

Pre-0.3.2, clients could expect to drop an UnboundedReceiver and observe
that incoming messages are dropped and the channel closed.

This change re-introduces this behavior for UnboundedReceiver and adds
tests to prevent regressions.

Note that a drop impl already exists for BoundedReceiver.

futures-channel/src/mpsc/mod.rs

@@ -1211,6 +1211,18 @@ impl<T> Stream for UnboundedReceiver<T> {
     }
 }
 
+impl<T> Drop for UnboundedReceiver<T> {
+    fn drop(&mut self) {
+        // Drain the channel of all pending messages
+        self.close();
+        if self.inner.is_some() {
+            while let Poll::Ready(Some(..)) = self.next_message() {
+                // ...
+            }
+        }
+    }
+}
+
 /*
  *
  * ===== impl Inner =====

futures-channel/tests/mpsc-close.rs

@@ -2,6 +2,7 @@ use futures::channel::mpsc;
 use futures::executor::block_on;
 use futures::sink::SinkExt;
 use futures::stream::StreamExt;
+use std::sync::Arc;
 use std::thread;
 
 #[test]
@@ -99,3 +100,45 @@ fn multiple_senders_close_channel() {
         assert_eq!(block_on(rx.next()), None);
     }
 }
+
+#[test]
+fn single_receiver_drop_closes_channel_and_drains() {
+    {
+        let ref_count = Arc::new(0);
+        let weak_ref = Arc::downgrade(&ref_count);
+
+        let (sender, receiver) = mpsc::unbounded();
+        sender.unbounded_send(ref_count).expect("failed to send");
+
+        // Verify that the sent message is still live.
+        assert!(weak_ref.upgrade().is_some());
+
+        drop(receiver);
+
+        // The sender should know the channel is closed.
+        assert!(sender.is_closed());
+
+        // Verify that the sent message has been dropped.
+        assert!(weak_ref.upgrade().is_none());
+    }
+
+    {
+        let ref_count = Arc::new(0);
+        let weak_ref = Arc::downgrade(&ref_count);
+
+        let (mut sender, receiver) = mpsc::channel(1);
+        sender.try_send(ref_count).expect("failed to send");
+
+        // Verify that the sent message is still live.
+        assert!(weak_ref.upgrade().is_some());
+
+        drop(receiver);
+
+        // The sender should know the channel is closed.
+        assert!(sender.is_closed());
+
+        // Verify that the sent message has been dropped.
+        assert!(weak_ref.upgrade().is_none());
+        assert!(sender.is_closed());
+    }
+}