Update http body

tower-h2/Cargo.toml

@@ -14,7 +14,7 @@ tower-service = { git = "http://github.com/tower-rs/tower", branch = "std-future
 tower-util = { git = "http://github.com/tower-rs/tower", branch = "std-future" }
 h2 = { git = "https://github.com/LucioFranco/h2", branch = "lucio/tower-h2-hack" }
 http = "0.1"
-http-body = { git =  "https://github.com/hyperium/http-body", branch = "std-future" }
+http-body = { git =  "https://github.com/hyperium/http-body", branch = "lucio/pin" }
 log = "0.4"
 
 [dev-dependencies]

tower-h2/examples/client.rs

@@ -3,8 +3,8 @@
 use http::Request;
 use std::task::{Context, Poll};
 use tokio::net::TcpStream;
-use tokio_buf::BufStream;
 use tower_h2::Connection;
+use std::pin::Pin;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -30,11 +30,14 @@ impl From<Vec<u8>> for Body {
     }
 }
 
-impl BufStream for Body {
-    type Item = std::io::Cursor<Vec<u8>>;
+impl http_body::Body for Body {
+    type Data = std::io::Cursor<Vec<u8>>;
     type Error = std::io::Error;
 
-    fn poll_buf(&mut self, _cx: &mut Context<'_>) -> Poll<Option<Result<Self::Item, Self::Error>>> {
+    fn poll_data(
+        mut self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
         if self.0.is_empty() {
             return None.into();
         }
@@ -46,4 +49,11 @@ impl BufStream for Body {
 
         Some(Ok(buf)).into()
     }
+
+    fn poll_trailers(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+        Ok(None).into()
+    }
 }

tower-h2/examples/server.rs

@@ -4,7 +4,7 @@ use futures_util::future;
 use http::{Request, Response};
 use std::task::{Context, Poll};
 use tokio::net::TcpListener;
-use tokio_buf::BufStream;
+use std::pin::Pin;
 use tower_h2::{RecvBody, Server};
 use tower_service::Service;
 
@@ -84,11 +84,14 @@ impl From<Vec<u8>> for Body {
     }
 }
 
-impl BufStream for Body {
-    type Item = std::io::Cursor<Vec<u8>>;
+impl http_body::Body for Body {
+    type Data = std::io::Cursor<Vec<u8>>;
     type Error = std::io::Error;
 
-    fn poll_buf(&mut self, _cx: &mut Context<'_>) -> Poll<Option<Result<Self::Item, Self::Error>>> {
+    fn poll_data(
+        mut self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
         if self.0.is_empty() {
             return None.into();
         }
@@ -100,4 +103,11 @@ impl BufStream for Body {
 
         Some(Ok(buf)).into()
     }
+
+    fn poll_trailers(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+        Ok(None).into()
+    }
 }

tower-h2/src/client.rs

@@ -57,10 +57,10 @@ where
     }
 
     fn call(&mut self, request: Request<B>) -> Self::Future {
-        let (parts, body) = request.into_parts();
+        let (parts, mut body) = request.into_parts();
         let request = Request::from_parts(parts, ());
 
-        let eos = body.is_end_stream();
+        let eos = Pin::new(&mut body).is_end_stream();
 
         let res = self.client.send_request(request, eos);
 

tower-h2/src/flush.rs

@@ -13,7 +13,7 @@ where
     S: Body,
 {
     h2: SendStream<SendBuf<S::Data>>,
-    body: S,
+    body: Pin<Box<dyn Body<Data = S::Data, Error = S::Error> + Send + 'static>>,
     state: FlushState,
 }
 
@@ -32,13 +32,13 @@ enum DataOrTrailers<B> {
 
 impl<S> Flush<S>
 where
-    S: Body,
+    S: Body + Send + 'static,
     S::Error: Into<Box<dyn std::error::Error>>,
 {
     pub fn new(src: S, dst: SendStream<SendBuf<S::Data>>) -> Self {
         Flush {
             h2: dst,
-            body: src,
+            body: Box::pin(src),
             state: FlushState::Data,
         }
     }
@@ -50,7 +50,7 @@ where
         loop {
             match ready!(self.poll_body(cx)) {
                 Some(Ok(Data(buf))) => {
-                    let eos = self.body.is_end_stream();
+                    let eos = Pin::new(&mut self.body).is_end_stream();
 
                     self.h2.send_data(SendBuf::new(buf), eos)?;
 
@@ -125,7 +125,7 @@ where
                         }
                     }
 
-                    let item = match ready!(self.body.poll_data(cx)) {
+                    let item = match ready!(Pin::new(&mut self.body).poll_data(cx)) {
                         Some(Ok(d)) => Some(d),
                         Some(Err(err)) => {
                             let err = err.into();
@@ -162,13 +162,14 @@ where
                             // before we get a RST_STREAM.
                         }
                     }
-                    let trailers = ready!(self.body.poll_trailers(cx).map_err(|err| {
-                        let err = err.into();
-                        debug!("user body error from poll_trailers: {}", err);
-                        let reason = crate::error::reason_from_dyn_error(&*err);
-                        self.h2.send_reset(reason);
-                        reason
-                    }))?;
+                    let trailers =
+                        ready!(Pin::new(&mut self.body).poll_trailers(cx).map_err(|err| {
+                            let err = err.into();
+                            debug!("user body error from poll_trailers: {}", err);
+                            let reason = crate::error::reason_from_dyn_error(&*err);
+                            self.h2.send_reset(reason);
+                            reason
+                        }))?;
                     self.state = FlushState::Done;
                     if let Some(trailers) = trailers {
                         return Some(Ok(DataOrTrailers::Trailers(trailers))).into();
@@ -182,7 +183,7 @@ where
 
 impl<S> Future for Flush<S>
 where
-    S: Body + Unpin,
+    S: Body + Send + 'static,
     S::Error: Into<Box<dyn std::error::Error>>,
 {
     type Output = Result<(), ()>;

tower-h2/src/recv_body.rs

@@ -1,6 +1,7 @@
 use bytes::{Buf, Bytes, BytesMut};
 use futures_util::TryStreamExt;
 use http_body::Body;
+use std::pin::Pin;
 use std::task::{Context, Poll};
 
 /// Allows a stream to be read from the remote.
@@ -33,11 +34,14 @@ impl Body for RecvBody {
     type Data = Data;
     type Error = h2::Error;
 
-    fn is_end_stream(&self) -> bool {
+    fn is_end_stream(self: Pin<&mut Self>) -> bool {
         self.inner.is_end_stream()
     }
 
-    fn poll_data(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Self::Data, h2::Error>>> {
+    fn poll_data(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Self::Data, h2::Error>>> {
         let data = match futures_util::ready!(self.inner.try_poll_next_unpin(cx)) {
             Some(Ok(bytes)) => {
                 self.inner
@@ -54,7 +58,7 @@ impl Body for RecvBody {
     }
 
     fn poll_trailers(
-        &mut self,
+        mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<Option<http::HeaderMap>, h2::Error>> {
         match futures_util::ready!(self.inner.poll_trailers(cx)) {

tower-h2/src/server.rs

@@ -86,10 +86,10 @@ pub async fn handle_request<B>(
     B::Data: Unpin,
     B::Error: Into<Box<dyn std::error::Error>>,
 {
-    let (parts, body) = response.into_parts();
+    let (parts, mut body) = response.into_parts();
 
     // Check if the response is imemdiately an end-of-stream.
-    let eos = body.is_end_stream();
+    let eos = std::pin::Pin::new(&mut body).is_end_stream();
 
     let response = Response::from_parts(parts, ());