fix(client): drop connection if body not consumed

client/src/body.rs

@@ -66,6 +66,17 @@ impl ResponseBody {
     }
 }
 
+impl Drop for ResponseBody {
+    fn drop(&mut self) {
+        #[cfg(feature = "http1")]
+        if let Self::H1(ref mut body) = *self {
+            if !body.is_consumed() {
+                body.conn_mut().destroy_on_drop()
+            }
+        }
+    }
+}
+
 impl Stream for ResponseBody {
     type Item = Result<Bytes, BodyError>;
 

client/src/h1/body.rs

@@ -31,6 +31,10 @@ impl<C> ResponseBody<C> {
     pub(crate) fn conn_mut(&mut self) -> &mut C {
         &mut self.conn
     }
+
+    pub(crate) fn is_consumed(&self) -> bool {
+        self.decoder.is_consumed()
+    }
 }
 
 impl<C> Stream for ResponseBody<C>

http/src/h1/proto/codec.rs

@@ -316,6 +316,14 @@ impl TransferCoding {
         }
     }
 
+    /// Determine if the body was consumed given the current state of the codec
+    pub fn is_consumed(&self) -> bool {
+        match self {
+            Self::Eof | Self::Length(0) | Self::DecodeChunked(ChunkedState::End, _) => true,
+            _ => false,
+        }
+    }
+
     /// decode body. See [ChunkResult] for detailed outcome.
     pub fn decode(&mut self, src: &mut BytesMut) -> ChunkResult {
         match *self {