fix impls of AsyncRead to not error when stream closed

Closes #12

this is also from cbaf050

Co-authored-by: Calvin Prewitt <[email protected]>

Author: pchickey

src/http/response.rs

@@ -117,12 +117,16 @@ impl AsyncRead for IncomingBody {
                 Reactor::current().wait_for(pollable).await;
 
                 // Read the bytes from the body stream
-                let buf = self.body_stream.read(CHUNK_SIZE).map_err(|err| match err {
-                    StreamError::LastOperationFailed(err) => {
-                        std::io::Error::other(format!("{}", err.to_debug_string()))
+                let buf = match self.body_stream.read(CHUNK_SIZE) {
+                    Ok(buf) => buf,
+                    Err(StreamError::Closed) => return Ok(0),
+                    Err(StreamError::LastOperationFailed(err)) => {
+                        return Err(std::io::Error::other(format!(
+                            "last operation failed: {}",
+                            err.to_debug_string()
+                        )))
                     }
-                    StreamError::Closed => std::io::Error::other("Connection closed"),
-                })?;
+                };
                 self.buf.insert(buf)
             }
         };

src/net/tcp_stream.rs

@@ -31,7 +31,11 @@ impl TcpStream {
 impl AsyncRead for TcpStream {
     async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         Reactor::current().wait_for(self.input.subscribe()).await;
-        let slice = self.input.read(buf.len() as u64).map_err(to_io_err)?;
+        let slice = match self.input.read(buf.len() as u64) {
+            Ok(slice) => slice,
+            Err(StreamError::Closed) => return Ok(0),
+            Err(e) => return Err(to_io_err(e)),
+        };
         let bytes_read = slice.len();
         buf[..bytes_read].clone_from_slice(&slice);
         Ok(bytes_read)
@@ -41,7 +45,11 @@ impl AsyncRead for TcpStream {
 impl AsyncRead for &TcpStream {
     async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         Reactor::current().wait_for(self.input.subscribe()).await;
-        let slice = self.input.read(buf.len() as u64).map_err(to_io_err)?;
+        let slice = match self.input.read(buf.len() as u64) {
+            Ok(slice) => slice,
+            Err(StreamError::Closed) => return Ok(0),
+            Err(e) => return Err(to_io_err(e)),
+        };
         let bytes_read = slice.len();
         buf[..bytes_read].clone_from_slice(&slice);
         Ok(bytes_read)