0.14 cherry-pick recent body errors fixes

Cargo.toml

@@ -62,7 +62,7 @@ tokio = { version = "1", features = [
 ] }
 tokio-test = "0.4"
 tokio-util = { version = "0.7", features = ["codec"] }
-tower = { version = "0.4", features = ["make", "util"] }
+tower = { version = "0.4", default-features = false, features = ["make", "util"] }
 url = "2.2"
 
 [target.'cfg(any(target_os = "linux", target_os = "macos"))'.dev-dependencies]

src/body/body.rs

@@ -602,17 +602,16 @@ impl Sender {
     }
 
     /// Aborts the body in an abnormal fashion.
-    pub fn abort(self) {
+    pub fn abort(mut self) {
+        self.send_error(crate::Error::new_body_write_aborted());
+    }
+
+    pub(crate) fn send_error(&mut self, err: crate::Error) {
         let _ = self
             .data_tx
             // clone so the send works even if buffer is full
             .clone()
-            .try_send(Err(crate::Error::new_body_write_aborted()));
-    }
-
-    #[cfg(feature = "http1")]
-    pub(crate) fn send_error(&mut self, err: crate::Error) {
-        let _ = self.data_tx.try_send(Err(err));
+            .try_send(Err(err));
     }
 }
 

src/proto/h1/dispatch.rs

@@ -118,6 +118,10 @@ where
         should_shutdown: bool,
     ) -> Poll<crate::Result<Dispatched>> {
         Poll::Ready(ready!(self.poll_inner(cx, should_shutdown)).or_else(|e| {
+            // Be sure to alert a streaming body of the failure.
+            if let Some(mut body) = self.body_tx.take() {
+                body.send_error(crate::Error::new_body("connection error"));
+            }
             // An error means we're shutting down either way.
             // We just try to give the error to the user,
             // and close the connection with an Ok. If we
@@ -367,7 +371,12 @@ where
                         self.conn.end_body()?;
                     }
                 } else {
-                    return Poll::Pending;
+                    // If there's no body_rx, end the body
+                    if self.conn.can_write_body() {
+                        self.conn.end_body()?;
+                    } else {
+                        return Poll::Pending;
+                    }
                 }
             }
         }

tests/server.rs

@@ -93,6 +93,7 @@ mod response_body_lengths {
     }
 
     fn run_test(case: TestCase) {
+        let _ = pretty_env_logger::try_init();
         assert!(
             case.version == 0 || case.version == 1,
             "TestCase.version must 0 or 1"
@@ -157,18 +158,22 @@ mod response_body_lengths {
         let n = body.find("\r\n\r\n").unwrap() + 4;
 
         if case.expects_chunked {
-            let len = body.len();
-            assert_eq!(
-                &body[n + 1..n + 3],
-                "\r\n",
-                "expected body chunk size header"
-            );
-            assert_eq!(&body[n + 3..len - 7], body_str, "expected body");
-            assert_eq!(
-                &body[len - 7..],
-                "\r\n0\r\n\r\n",
-                "expected body final chunk size header"
-            );
+            if body_str.len() > 0 {
+                let len = body.len();
+                assert_eq!(
+                    &body[n + 1..n + 3],
+                    "\r\n",
+                    "expected body chunk size header"
+                );
+                assert_eq!(&body[n + 3..len - 7], body_str, "expected body");
+                assert_eq!(
+                    &body[len - 7..],
+                    "\r\n0\r\n\r\n",
+                    "expected body final chunk size header"
+                );
+            } else {
+                assert_eq!(&body[n..], "0\r\n\r\n");
+            }
         } else {
             assert_eq!(&body[n..], body_str, "expected body");
         }
@@ -219,6 +224,17 @@ mod response_body_lengths {
         });
     }
 
+    #[test]
+    fn chunked_response_known_empty() {
+        run_test(TestCase {
+            version: 1,
+            headers: &[("transfer-encoding", "chunked")],
+            body: Bd::Known(""),
+            expects_chunked: true, // should still send chunked, and 0\r\n\r\n
+            expects_con_len: false,
+        });
+    }
+
     #[test]
     fn chunked_response_unknown() {
         run_test(TestCase {