fix(client): fix connection leak when Response finishes before Request body

If the Response was received and the body finished while the Request
body was still streaming, the connection could get into a state where it
was never polled again, thus never re-inserting into the connection pool
or being dropped.

Closes #1717

src/proto/h1/conn.rs

@@ -794,7 +794,7 @@ impl State {
         match (&self.reading, &self.writing) {
             (&Reading::KeepAlive, &Writing::KeepAlive) => {
                 if let KA::Busy = self.keep_alive.status() {
-                    self.idle();
+                    self.idle::<T>();
                 } else {
                     trace!("try_keep_alive({}): could keep-alive, but status = {:?}", T::LOG, self.keep_alive);
                     self.close();
@@ -819,12 +819,23 @@ impl State {
         self.keep_alive.busy();
     }
 
-    fn idle(&mut self) {
+    fn idle<T: Http1Transaction>(&mut self) {
+        debug_assert!(!self.is_idle(), "State::idle() called while idle");
+
         self.method = None;
         self.keep_alive.idle();
         if self.is_idle() {
             self.reading = Reading::Init;
             self.writing = Writing::Init;
+
+            // !T::should_read_first() means Client.
+            //
+            // If Client connection has just gone idle, the Dispatcher
+            // should try the poll loop one more time, so as to poll the
+            // pending requests stream.
+            if !T::should_read_first() {
+                self.notify_read = true;
+            }
         } else {
             self.close();
         }

src/proto/h1/dispatch.rs

@@ -466,7 +466,7 @@ where
                 // user has dropped sender handle
                 Ok(Async::Ready(None))
             },
-            Ok(Async::NotReady) => return Ok(Async::NotReady),
+            Ok(Async::NotReady) => Ok(Async::NotReady),
             Err(never) => match never {},
         }
     }

tests/client.rs

@@ -1215,6 +1215,83 @@ mod dispatch_impl {
         assert_eq!(connects.load(Ordering::Relaxed), 2);
     }
 
+    #[test]
+    fn client_keep_alive_when_response_before_request_body_ends() {
+        use futures_timer::Delay;
+        let _ = pretty_env_logger::try_init();
+        let server = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = server.local_addr().unwrap();
+        let mut rt = Runtime::new().unwrap();
+
+        let connector = DebugConnector::new();
+        let connects = connector.connects.clone();
+
+        let client = Client::builder()
+            .build(connector);
+
+        let (tx1, rx1) = oneshot::channel();
+        let (tx2, rx2) = oneshot::channel();
+        let (tx3, rx3) = oneshot::channel();
+        thread::spawn(move || {
+            let mut sock = server.accept().unwrap().0;
+            sock.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+            sock.set_write_timeout(Some(Duration::from_secs(5))).unwrap();
+            let mut buf = [0; 4096];
+            sock.read(&mut buf).expect("read 1");
+            sock.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n").expect("write 1");
+            // after writing the response, THEN stream the body
+            let _ = tx1.send(());
+
+            sock.read(&mut buf).expect("read 2");
+            let _ = tx2.send(());
+
+            let n2 = sock.read(&mut buf).expect("read 3");
+            assert_ne!(n2, 0);
+            let second_get = "GET /b HTTP/1.1\r\n";
+            assert_eq!(s(&buf[..second_get.len()]), second_get);
+            sock.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n").expect("write 2");
+            let _ = tx3.send(());
+        });
+
+
+        assert_eq!(connects.load(Ordering::Relaxed), 0);
+
+        let delayed_body = rx1
+            .map_err(|_| -> hyper::Error { panic!("rx1") })
+            .and_then(|_| Delay::new(Duration::from_millis(200)).map_err(|_| panic!("delay")))
+            .into_stream()
+            .map(|_| "hello a");
+
+        let rx = rx2.expect("thread panicked");
+        let req = Request::builder()
+            .method("POST")
+            .uri(&*format!("http://{}/a", addr))
+            .body(Body::wrap_stream(delayed_body))
+            .unwrap();
+        let client2 = client.clone();
+
+        // req 1
+        let fut = client.request(req)
+            .join(rx)
+            .and_then(|_| Delay::new(Duration::from_millis(200)).expect("delay"))
+            // req 2
+            .and_then(move |()| {
+                let rx = rx3.expect("thread panicked");
+                let req = Request::builder()
+                    .uri(&*format!("http://{}/b", addr))
+                    .body(Body::empty())
+                    .unwrap();
+                client2
+                    .request(req)
+                    .join(rx)
+                    .map(|_| ())
+            });
+
+        rt.block_on(fut).unwrap();
+
+        assert_eq!(connects.load(Ordering::Relaxed), 1);
+    }
+
     #[test]
     fn connect_proxy_sends_absolute_uri() {
         let _ = pretty_env_logger::try_init();