feat(body): add trailers to Body channel (#2260)

benches/end_to_end.rs

@@ -8,7 +8,7 @@ use std::net::SocketAddr;
 use futures_util::future::join_all;
 
 use hyper::client::HttpConnector;
-use hyper::{body::HttpBody as _, Body, HeaderMap, Method, Request, Response, Server};
+use hyper::{body::HttpBody as _, Body, Method, Request, Response, Server};
 
 // HTTP1
 
@@ -313,7 +313,6 @@ impl Opts {
                     for _ in 0..chunk_cnt {
                         tx.send_data(chunk.into()).await.expect("send_data");
                     }
-                    tx.send_trailers(HeaderMap::new()).expect("send_trailers");
                 });
                 body
             } else {

src/body/body.rs

@@ -382,7 +382,7 @@ impl HttpBody for Body {
                 ..
             } => match ready!(Pin::new(trailers_rx).poll(cx)) {
                 Ok(t) => Poll::Ready(Ok(Some(t))),
-                Err(_) => Poll::Ready(Err(crate::Error::new_closed())),
+                Err(_) => Poll::Ready(Ok(None)),
             },
             #[cfg(feature = "ffi")]
             Kind::Ffi(ref mut body) => body.poll_trailers(cx),
@@ -568,7 +568,7 @@ impl Sender {
     }
 
     /// Send trailers on trailers channel.
-    pub fn send_trailers(&mut self, trailers: HeaderMap) -> crate::Result<()> {
+    pub async fn send_trailers(&mut self, trailers: HeaderMap) -> crate::Result<()> {
         let tx = match self.trailers_tx.take() {
             Some(tx) => tx,
             None => return Err(crate::Error::new_closed()),

src/client/conn.rs

@@ -63,10 +63,7 @@ use tower_service::Service;
 
 use super::dispatch;
 use crate::body::HttpBody;
-use crate::common::{
-    exec::{BoxSendFuture, Exec},
-    task, Future, Pin, Poll,
-};
+use crate::common::{task, exec::{BoxSendFuture, Exec}, Future, Pin, Poll};
 use crate::proto;
 use crate::rt::Executor;
 #[cfg(feature = "http1")]