Update to hyper 1.0-rc.3 and http-body-util 0.1.0-rc.2

axum-core/src/body.rs

@@ -2,17 +2,16 @@
 
 use crate::{BoxError, Error};
 use bytes::Bytes;
-use bytes::{Buf, BufMut};
 use futures_util::stream::Stream;
 use futures_util::TryStream;
-use http::HeaderMap;
-use http_body::Body as _;
+use http_body::{Body as _, Frame};
+use http_body_util::BodyExt;
 use pin_project_lite::pin_project;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use sync_wrapper::SyncWrapper;
 
-type BoxBody = http_body::combinators::UnsyncBoxBody<Bytes, Error>;
+type BoxBody = http_body_util::combinators::UnsyncBoxBody<Bytes, Error>;
 
 fn boxed<B>(body: B) -> BoxBody
 where
@@ -35,58 +34,6 @@ where
     }
 }
 
-// copied from hyper under the following license:
-// Copyright (c) 2014-2021 Sean McArthur
-
-// Permission is hereby granted, free of charge, to any person obtaining a copy
-// of this software and associated documentation files (the "Software"), to deal
-// in the Software without restriction, including without limitation the rights
-// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-// copies of the Software, and to permit persons to whom the Software is
-// furnished to do so, subject to the following conditions:
-
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
-
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
-pub(crate) async fn to_bytes<T>(body: T) -> Result<Bytes, T::Error>
-where
-    T: http_body::Body,
-{
-    futures_util::pin_mut!(body);
-
-    // If there's only 1 chunk, we can just return Buf::to_bytes()
-    let mut first = if let Some(buf) = body.data().await {
-        buf?
-    } else {
-        return Ok(Bytes::new());
-    };
-
-    let second = if let Some(buf) = body.data().await {
-        buf?
-    } else {
-        return Ok(first.copy_to_bytes(first.remaining()));
-    };
-
-    // With more than 1 buf, we gotta flatten into a Vec first.
-    let cap = first.remaining() + second.remaining() + body.size_hint().lower() as usize;
-    let mut vec = Vec::with_capacity(cap);
-    vec.put(first);
-    vec.put(second);
-
-    while let Some(buf) = body.data().await {
-        vec.put(buf?);
-    }
-
-    Ok(vec.into())
-}
-
 /// The body type used in axum requests and responses.
 #[derive(Debug)]
 pub struct Body(BoxBody);
@@ -103,7 +50,7 @@ impl Body {
 
     /// Create an empty body.
     pub fn empty() -> Self {
-        Self::new(http_body::Empty::new())
+        Self::new(http_body_util::Empty::new())
     }
 
     /// Create a new `Body` from a [`Stream`].
@@ -131,7 +78,7 @@ macro_rules! body_from_impl {
     ($ty:ty) => {
         impl From<$ty> for Body {
             fn from(buf: $ty) -> Self {
-                Self::new(http_body::Full::from(buf))
+                Self::new(http_body_util::Full::from(buf))
             }
         }
     };
@@ -152,19 +99,11 @@ impl http_body::Body for Body {
     type Error = Error;
 
     #[inline]
-    fn poll_data(
+    fn poll_frame(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> std::task::Poll<Option<Result<Self::Data, Self::Error>>> {
-        Pin::new(&mut self.0).poll_data(cx)
-    }
-
-    #[inline]
-    fn poll_trailers(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> std::task::Poll<Result<Option<HeaderMap>, Self::Error>> {
-        Pin::new(&mut self.0).poll_trailers(cx)
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        Pin::new(&mut self.0).poll_frame(cx)
     }
 
     #[inline]
@@ -182,8 +121,16 @@ impl Stream for Body {
     type Item = Result<Bytes, Error>;
 
     #[inline]
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.poll_data(cx)
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        loop {
+            match futures_util::ready!(self.as_mut().poll_frame(cx)?) {
+                Some(frame) => match frame.into_data() {
+                    Ok(data) => return Poll::Ready(Some(Ok(data))),
+                    Err(_frame) => {}
+                },
+                None => return Poll::Ready(None),
+            }
+        }
     }
 }
 
@@ -203,25 +150,17 @@ where
     type Data = Bytes;
     type Error = Error;
 
-    fn poll_data(
+    fn poll_frame(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
         let stream = self.project().stream.get_pin_mut();
         match futures_util::ready!(stream.try_poll_next(cx)) {
-            Some(Ok(chunk)) => Poll::Ready(Some(Ok(chunk.into()))),
+            Some(Ok(chunk)) => Poll::Ready(Some(Ok(Frame::data(chunk.into())))),
             Some(Err(err)) => Poll::Ready(Some(Err(Error::new(err)))),
             None => Poll::Ready(None),
         }
     }
-
-    #[inline]
-    fn poll_trailers(
-        self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
-        Poll::Ready(Ok(None))
-    }
 }
 
 #[test]

axum-core/src/ext_traits/request.rs

@@ -1,7 +1,7 @@
 use crate::body::Body;
 use crate::extract::{DefaultBodyLimitKind, FromRequest, FromRequestParts, Request};
 use futures_util::future::BoxFuture;
-use http_body::Limited;
+use http_body_util::Limited;
 
 mod sealed {
     pub trait Sealed {}
@@ -326,9 +326,9 @@ impl RequestExt for Request {
         match self.extensions().get::<DefaultBodyLimitKind>().copied() {
             Some(DefaultBodyLimitKind::Disable) => Err(self),
             Some(DefaultBodyLimitKind::Limit(limit)) => {
-                Ok(self.map(|b| http_body::Limited::new(b, limit)))
+                Ok(self.map(|b| http_body_util::Limited::new(b, limit)))
             }
-            None => Ok(self.map(|b| http_body::Limited::new(b, DEFAULT_LIMIT))),
+            None => Ok(self.map(|b| http_body_util::Limited::new(b, DEFAULT_LIMIT))),
         }
     }
 

axum-core/src/extract/rejection.rs

@@ -20,7 +20,7 @@ impl FailedToBufferBody {
             Ok(err) => err.into_inner(),
             Err(err) => err,
         };
-        match box_error.downcast::<http_body::LengthLimitError>() {
+        match box_error.downcast::<http_body_util::LengthLimitError>() {
             Ok(err) => Self::LengthLimitError(LengthLimitError::from_err(err)),
             Err(err) => Self::UnknownBodyError(UnknownBodyError::from_err(err)),
         }

axum-core/src/extract/request_parts.rs

@@ -3,6 +3,7 @@ use crate::{body::Body, RequestExt};
 use async_trait::async_trait;
 use bytes::Bytes;
 use http::{request::Parts, HeaderMap, Method, Uri, Version};
+use http_body_util::BodyExt;
 use std::convert::Infallible;
 
 #[async_trait]
@@ -79,12 +80,16 @@ where
 
     async fn from_request(req: Request, _: &S) -> Result<Self, Self::Rejection> {
         let bytes = match req.into_limited_body() {
-            Ok(limited_body) => crate::body::to_bytes(limited_body)
+            Ok(limited_body) => limited_body
+                .collect()
                 .await
-                .map_err(FailedToBufferBody::from_err)?,
-            Err(unlimited_body) => crate::body::to_bytes(unlimited_body)
+                .map_err(FailedToBufferBody::from_err)?
+                .to_bytes(),
+            Err(unlimited_body) => unlimited_body
+                .collect()
                 .await
-                .map_err(FailedToBufferBody::from_err)?,
+                .map_err(FailedToBufferBody::from_err)?
+                .to_bytes(),
         };
 
         Ok(bytes)

axum-core/src/response/into_response.rs

@@ -5,7 +5,7 @@ use http::{
     header::{self, HeaderMap, HeaderName, HeaderValue},
     Extensions, StatusCode,
 };
-use http_body::SizeHint;
+use http_body::{Frame, SizeHint};
 use std::{
     borrow::Cow,
     convert::Infallible,
@@ -250,30 +250,23 @@ where
     type Data = Bytes;
     type Error = Infallible;
 
-    fn poll_data(
+    fn poll_frame(
         mut self: Pin<&mut Self>,
         _cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
         if let Some(mut buf) = self.first.take() {
             let bytes = buf.copy_to_bytes(buf.remaining());
-            return Poll::Ready(Some(Ok(bytes)));
+            return Poll::Ready(Some(Ok(Frame::data(bytes))));
         }
 
         if let Some(mut buf) = self.second.take() {
             let bytes = buf.copy_to_bytes(buf.remaining());
-            return Poll::Ready(Some(Ok(bytes)));
+            return Poll::Ready(Some(Ok(Frame::data(bytes))));
         }
 
         Poll::Ready(None)
     }
 
-    fn poll_trailers(
-        self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
-        Poll::Ready(Ok(None))
-    }
-
     fn is_end_stream(&self) -> bool {
         self.first.is_none() && self.second.is_none()
     }