Add body combinators (#36)

* Add body combinators

Adds `BodyExt` with the following methods

- `map_data`
- `map_err`
- `boxed`

Disabled by default and can be enabled with the "util" feature.

Fixes #34

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Unreleased
 
+- Add combinators to `Body`:
+  - `map_data`: Change the `Data` chunks produced by the body.
+  - `map_err`: Change the `Error`s produced by the body.
+  - `boxed`: Convert the `Body` into a boxed trait object.
 - Add `Empty`.
 
 # 0.4.0 (December 23, 2020)

Cargo.toml

@@ -28,3 +28,4 @@ categories = ["web-programming"]
 [dependencies]
 bytes = "1"
 http = "0.2"
+pin-project-lite = "0.2"

src/combinators/box_body.rs

@@ -0,0 +1,61 @@
+use crate::Body;
+use bytes::Buf;
+use std::{
+    fmt,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+/// A boxed [`Body`] trait object.
+pub struct BoxBody<D, E> {
+    inner: Pin<Box<dyn Body<Data = D, Error = E> + Send + Sync + 'static>>,
+}
+
+impl<D, E> BoxBody<D, E> {
+    /// Create a new `BoxBody`.
+    pub fn new<B>(body: B) -> Self
+    where
+        B: Body<Data = D, Error = E> + Send + Sync + 'static,
+        D: Buf,
+    {
+        Self {
+            inner: Box::pin(body),
+        }
+    }
+}
+
+impl<D, E> fmt::Debug for BoxBody<D, E> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("BoxBody").finish()
+    }
+}
+
+impl<D, E> Body for BoxBody<D, E>
+where
+    D: Buf,
+{
+    type Data = D;
+    type Error = E;
+
+    fn poll_data(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+        self.inner.as_mut().poll_data(cx)
+    }
+
+    fn poll_trailers(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+        self.inner.as_mut().poll_trailers(cx)
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+
+    fn size_hint(&self) -> crate::SizeHint {
+        self.inner.size_hint()
+    }
+}

src/combinators/map_data.rs

@@ -0,0 +1,80 @@
+use crate::Body;
+use bytes::Buf;
+use pin_project_lite::pin_project;
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+pin_project! {
+    /// Body returned by the [`map_data`] combinator.
+    ///
+    /// [`map_data`]: crate::util::BodyExt::map_data
+    #[derive(Debug, Clone, Copy)]
+    pub struct MapData<B, F> {
+        #[pin]
+        inner: B,
+        f: F
+    }
+}
+
+impl<B, F> MapData<B, F> {
+    #[inline]
+    pub(crate) fn new(body: B, f: F) -> Self {
+        Self { inner: body, f }
+    }
+
+    /// Get a reference to the inner body
+    pub fn get_ref(&self) -> &B {
+        &self.inner
+    }
+
+    /// Get a mutable reference to the inner body
+    pub fn get_mut(&mut self) -> &mut B {
+        &mut self.inner
+    }
+
+    /// Get a pinned mutable reference to the inner body
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut B> {
+        self.project().inner
+    }
+
+    /// Consume `self`, returning the inner body
+    pub fn into_inner(self) -> B {
+        self.inner
+    }
+}
+
+impl<B, F, B2> Body for MapData<B, F>
+where
+    B: Body,
+    F: FnMut(B::Data) -> B2,
+    B2: Buf,
+{
+    type Data = B2;
+    type Error = B::Error;
+
+    fn poll_data(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+        let this = self.project();
+        match this.inner.poll_data(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Ready(Some(Ok(data))) => Poll::Ready(Some(Ok((this.f)(data)))),
+            Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(err))),
+        }
+    }
+
+    fn poll_trailers(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+        self.project().inner.poll_trailers(cx)
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+}

src/combinators/map_err.rs

@@ -0,0 +1,83 @@
+use crate::Body;
+use pin_project_lite::pin_project;
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+pin_project! {
+    /// Body returned by the [`map_err`] combinator.
+    ///
+    /// [`map_err`]: crate::util::BodyExt::map_err
+    #[derive(Debug, Clone, Copy)]
+    pub struct MapErr<B, F> {
+        #[pin]
+        inner: B,
+        f: F
+    }
+}
+
+impl<B, F> MapErr<B, F> {
+    #[inline]
+    pub(crate) fn new(body: B, f: F) -> Self {
+        Self { inner: body, f }
+    }
+
+    /// Get a reference to the inner body
+    pub fn get_ref(&self) -> &B {
+        &self.inner
+    }
+
+    /// Get a mutable reference to the inner body
+    pub fn get_mut(&mut self) -> &mut B {
+        &mut self.inner
+    }
+
+    /// Get a pinned mutable reference to the inner body
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut B> {
+        self.project().inner
+    }
+
+    /// Consume `self`, returning the inner body
+    pub fn into_inner(self) -> B {
+        self.inner
+    }
+}
+
+impl<B, F, E> Body for MapErr<B, F>
+where
+    B: Body,
+    F: FnMut(B::Error) -> E,
+{
+    type Data = B::Data;
+    type Error = E;
+
+    fn poll_data(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+        let this = self.project();
+        match this.inner.poll_data(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Ready(Some(Ok(data))) => Poll::Ready(Some(Ok(data))),
+            Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err((this.f)(err)))),
+        }
+    }
+
+    fn poll_trailers(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+        let this = self.project();
+        this.inner.poll_trailers(cx).map_err(this.f)
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+
+    fn size_hint(&self) -> crate::SizeHint {
+        self.inner.size_hint()
+    }
+}

src/combinators/mod.rs

@@ -0,0 +1,7 @@
+//! Combinators for the `Body` trait.
+
+mod box_body;
+mod map_data;
+mod map_err;
+
+pub use self::{box_body::BoxBody, map_data::MapData, map_err::MapErr};

src/lib.rs

@@ -1,5 +1,10 @@
 #![doc(html_root_url = "https://docs.rs/http-body/0.4.0")]
-#![deny(missing_debug_implementations, missing_docs, unreachable_pub)]
+#![deny(
+    missing_debug_implementations,
+    missing_docs,
+    unreachable_pub,
+    broken_intra_doc_links
+)]
 #![cfg_attr(test, deny(warnings))]
 
 //! Asynchronous HTTP request or response body.
@@ -12,10 +17,13 @@ mod empty;
 mod next;
 mod size_hint;
 
+pub mod combinators;
+
 pub use self::empty::Empty;
 pub use self::next::{Data, Trailers};
 pub use self::size_hint::SizeHint;
 
+use self::combinators::{BoxBody, MapData, MapErr};
 use bytes::Buf;
 use http::HeaderMap;
 use std::ops;
@@ -85,6 +93,33 @@ pub trait Body {
     {
         Trailers(self)
     }
+
+    /// Maps this body's data value to a different value.
+    fn map_data<F, B>(self, f: F) -> MapData<Self, F>
+    where
+        Self: Sized,
+        F: FnMut(Self::Data) -> B,
+        B: Buf,
+    {
+        MapData::new(self, f)
+    }
+
+    /// Maps this body's error value to a different value.
+    fn map_err<F, E>(self, f: F) -> MapErr<Self, F>
+    where
+        Self: Sized,
+        F: FnMut(Self::Error) -> E,
+    {
+        MapErr::new(self, f)
+    }
+
+    /// Turn this body into a boxed trait object.
+    fn boxed(self) -> BoxBody<Self::Data, Self::Error>
+    where
+        Self: Sized + Send + Sync + 'static,
+    {
+        BoxBody::new(self)
+    }
 }
 
 impl<T: Body + Unpin + ?Sized> Body for &mut T {