fix(client): send an Error::Cancel if a queued request is dropped

Adds `Error::Cancel` variant.

src/client/cancel.rs

@@ -57,6 +57,12 @@ impl Cancel {
     }
 }
 
+impl Canceled {
+    pub fn cancel(&self) {
+        self.inner.is_canceled.store(true, Ordering::SeqCst);
+    }
+}
+
 impl Future for Canceled {
     type Item = ();
     type Error = Never;
@@ -87,7 +93,7 @@ impl Future for Canceled {
 
 impl Drop for Canceled {
     fn drop(&mut self) {
-        self.inner.is_canceled.store(true, Ordering::SeqCst);
+        self.cancel();
     }
 }
 

src/client/dispatch.rs

@@ -69,21 +69,66 @@ impl<T, U> Stream for Receiver<T, U> {
     }
 }
 
-//TODO: Drop for Receiver should consume inner
+impl<T, U> Drop for Receiver<T, U> {
+    fn drop(&mut self) {
+        self.canceled.cancel();
+        self.inner.close();
+
+        // This poll() is safe to call in `Drop`, because we've
+        // called, `close`, which promises that no new messages
+        // will arrive, and thus, once we reach the end, we won't
+        // see a `NotReady` (and try to park), but a Ready(None).
+        //
+        // All other variants:
+        // - Ready(None): the end. we want to stop looping
+        // - NotReady: unreachable
+        // - Err: unreachable
+        while let Ok(Async::Ready(Some((_val, cb)))) = self.inner.poll() {
+            // maybe in future, we pass the value along with the error?
+            let _ = cb.send(Err(::Error::new_canceled()));
+        }
+    }
+
+}
 
 #[cfg(test)]
 mod tests {
-
+    extern crate pretty_env_logger;
     #[cfg(feature = "nightly")]
     extern crate test;
 
+    use futures::{future, Future};
+
     #[cfg(feature = "nightly")]
-    use futures::{Future, Stream};
+    use futures::{Stream};
+
+    #[test]
+    fn drop_receiver_sends_cancel_errors() {
+        let _ = pretty_env_logger::try_init();
+
+        future::lazy(|| {
+            #[derive(Debug)]
+            struct Custom(i32);
+            let (tx, rx) = super::channel::<Custom, ()>();
+
+            let promise = tx.send(Custom(43)).unwrap();
+            drop(rx);
+
+            promise.then(|fulfilled| {
+                let res = fulfilled.expect("fulfilled");
+                match res.unwrap_err() {
+                    ::Error::Cancel(_) => (),
+                    e => panic!("expected Error::Cancel(_), found {:?}", e),
+                }
+
+                Ok::<(), ()>(())
+            })
+        }).wait().unwrap();
+    }
 
     #[cfg(feature = "nightly")]
     #[bench]
     fn cancelable_queue_throughput(b: &mut test::Bencher) {
-
         let (tx, mut rx) = super::channel::<i32, ()>();
 
         b.iter(move || {