fix(http1): send error on Incoming body when connection errors (#3256)

If a connection has any error besides reading, a streaming body
sometimes wouldn't be notified. This change makes it so that when a
connection task is closing because of any error, an existing body
channel is also notified.

Closes #3253

src/body/incoming.rs

@@ -337,19 +337,17 @@ impl Sender {
             .map_err(|err| err.into_inner().expect("just sent Ok"))
     }
 
-    /// Aborts the body in an abnormal fashion.
     #[allow(unused)]
-    pub(crate) fn abort(self) {
+    pub(crate) 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

@@ -117,6 +117,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