fix(http1): reduce closed connections when body is dropped

If a user makes use of `Body::is_end_stream` to optimize so as to not
need to do make a final poll just to receive `None`, previously the
connection would not have progressed its reading state to a finished
body, and so the connection would be closed.

Now, upon reading any chunk, the connection state will check if it
can know that the body would be finished, and progresses to a body
finished state sooner.

The integration tests were amplified by adding a naive hyper proxy
as a secondary test, which happens to make use of that optimization,
and thus caught the issue.

src/client/tests.rs

@@ -137,7 +137,14 @@ fn checkout_win_allows_connect_future_to_be_pooled() {
             .map(|res| res.into_body().concat2());
         let srv1 = poll_fn(|| {
             try_ready!(sock1.read(&mut [0u8; 512]));
-            try_ready!(sock1.write(b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\nx"));
+            // Chunked is used so as to force 2 body reads.
+            try_ready!(sock1.write(b"\
+                HTTP/1.1 200 OK\r\n\
+                transfer-encoding: chunked\r\n\
+                \r\n\
+                1\r\nx\r\n\
+                0\r\n\r\n\
+            "));
             Ok(Async::Ready(()))
         }).map_err(|e: ::std::io::Error| panic!("srv1 poll_fn error: {}", e));
 

src/proto/h1/conn.rs

@@ -180,29 +180,31 @@ where I: AsyncRead + AsyncWrite,
     pub fn read_body(&mut self) -> Poll<Option<Chunk>, io::Error> {
         debug_assert!(self.can_read_body());
 
-        trace!("Conn::read_body");
-
         let (reading, ret) = match self.state.reading {
             Reading::Body(ref mut decoder) => {
                 match decoder.decode(&mut self.io) {
                     Ok(Async::Ready(slice)) => {
-                        let (reading, chunk) = if !slice.is_empty() {
-                            return Ok(Async::Ready(Some(Chunk::from(slice))));
-                        } else if decoder.is_eof() {
+                        let (reading, chunk) = if decoder.is_eof() {
                             debug!("incoming body completed");
-                            (Reading::KeepAlive, None)
-                        } else {
-                            trace!("decode stream unexpectedly ended");
-                            // this should actually be unreachable:
-                            // the decoder will return an UnexpectedEof if there were
-                            // no bytes to read and it isn't eof yet...
+                            (Reading::KeepAlive, if !slice.is_empty() {
+                                Some(Chunk::from(slice))
+                            } else {
+                                None
+                            })
+                        } else if slice.is_empty() {
+                            error!("decode stream unexpectedly ended");
+                            // This should be unreachable, since all 3 decoders
+                            // either set eof=true or return an Err when reading
+                            // an empty slice...
                             (Reading::Closed, None)
+                        } else {
+                            return Ok(Async::Ready(Some(Chunk::from(slice))));
                         };
                         (reading, Ok(Async::Ready(chunk)))
                     },
                     Ok(Async::NotReady) => return Ok(Async::NotReady),
                     Err(e) => {
-                        trace!("decode stream error: {}", e);
+                        debug!("decode stream error: {}", e);
                         (Reading::Closed, Err(e))
                     },
                 }