From 05830ab0b7a2a4240b5ce09103579272527aa528 Mon Sep 17 00:00:00 2001
From: "Stephen M. Coakley" <me@stephencoakley.com>
Date: Wed, 5 Jun 2019 22:47:45 -0500
Subject: [PATCH] fix: Reader should be drainable after writer disconnects

Fix an issue where dropping the writer while some data still remained in the pipe caused the reader to return an error instead of allowing you to drain the bytes in the pipe until empty.
---
 src/pipe/chunked.rs | 31 +++++++++++++++++++++++++++++--
 1 file changed, 29 insertions(+), 2 deletions(-)

diff --git a/src/pipe/chunked.rs b/src/pipe/chunked.rs
index c0ed1ad..d248ff5 100644
--- a/src/pipe/chunked.rs
+++ b/src/pipe/chunked.rs
@@ -117,7 +117,17 @@ impl AsyncRead for Reader {
                     // of buffers, so this can never happen.
                     if e.is_full() {
                         panic!("buffer pool overflow")
-                    } else {
+                    }
+
+                    // If the writer disconnects, then we'll just discard this
+                    // buffer and any subsequent buffers until we've read
+                    // everything still in the pipe.
+                    else if e.is_disconnected() {
+                        // Nothing!
+                    }
+
+                    // Some other error occurred.
+                    else {
                         return Poll::Ready(Err(io::ErrorKind::BrokenPipe.into()));
                     }
                 }
@@ -193,6 +203,23 @@ mod tests {
             let mut dest = [0; 5];
             assert_eq!(reader.read(&mut dest).await.unwrap(), 5);
             assert_eq!(&dest, b"hello");
-        });
+        })
+    }
+
+    #[test]
+    fn reader_still_drainable_after_writer_disconnects() {
+        block_on(async {
+            let (mut reader, mut writer) = new(1);
+
+            writer.write_all(b"hello").await.unwrap();
+
+            drop(writer);
+
+            let mut dest = [0; 5];
+            assert_eq!(reader.read(&mut dest).await.unwrap(), 5);
+            assert_eq!(&dest, b"hello");
+
+            assert_eq!(reader.read(&mut dest).await.unwrap(), 0);
+        })
     }
 }