http_body_util/
lib.rs

1#![deny(missing_debug_implementations, missing_docs, unreachable_pub)]
2#![cfg_attr(test, deny(warnings))]
3#![cfg_attr(docsrs, feature(doc_auto_cfg))]
4
5//! Utilities for [`http_body::Body`].
6//!
7//! [`BodyExt`] adds extensions to the common trait.
8//!
9//! [`Empty`] and [`Full`] provide simple implementations.
10
11mod collected;
12pub mod combinators;
13mod either;
14mod empty;
15mod full;
16mod limited;
17mod stream;
18
19#[cfg(feature = "channel")]
20pub mod channel;
21
22mod util;
23
24use self::combinators::{BoxBody, MapErr, MapFrame, UnsyncBoxBody};
25
26pub use self::collected::Collected;
27pub use self::either::Either;
28pub use self::empty::Empty;
29pub use self::full::Full;
30pub use self::limited::{LengthLimitError, Limited};
31pub use self::stream::{BodyDataStream, BodyStream, StreamBody};
32
33#[cfg(feature = "channel")]
34pub use self::channel::Channel;
35
36/// An extension trait for [`http_body::Body`] adding various combinators and adapters
37pub trait BodyExt: http_body::Body {
38    /// Returns a future that resolves to the next [`Frame`], if any.
39    ///
40    /// [`Frame`]: combinators::Frame
41    fn frame(&mut self) -> combinators::Frame<'_, Self>
42    where
43        Self: Unpin,
44    {
45        combinators::Frame(self)
46    }
47
48    /// Maps this body's frame to a different kind.
49    fn map_frame<F, B>(self, f: F) -> MapFrame<Self, F>
50    where
51        Self: Sized,
52        F: FnMut(http_body::Frame<Self::Data>) -> http_body::Frame<B>,
53        B: bytes::Buf,
54    {
55        MapFrame::new(self, f)
56    }
57
58    /// A body that calls a function with a reference to each frame before yielding it.
59    fn inspect_frame<F>(self, f: F) -> combinators::InspectFrame<Self, F>
60    where
61        Self: Sized,
62        F: FnMut(&http_body::Frame<Self::Data>),
63    {
64        combinators::InspectFrame::new(self, f)
65    }
66
67    /// Maps this body's error value to a different value.
68    fn map_err<F, E>(self, f: F) -> MapErr<Self, F>
69    where
70        Self: Sized,
71        F: FnMut(Self::Error) -> E,
72    {
73        MapErr::new(self, f)
74    }
75
76    /// A body that calls a function with a reference to an error before yielding it.
77    fn inspect_err<F>(self, f: F) -> combinators::InspectErr<Self, F>
78    where
79        Self: Sized,
80        F: FnMut(&Self::Error),
81    {
82        combinators::InspectErr::new(self, f)
83    }
84
85    /// Turn this body into a boxed trait object.
86    fn boxed(self) -> BoxBody<Self::Data, Self::Error>
87    where
88        Self: Sized + Send + Sync + 'static,
89    {
90        BoxBody::new(self)
91    }
92
93    /// Turn this body into a boxed trait object that is !Sync.
94    fn boxed_unsync(self) -> UnsyncBoxBody<Self::Data, Self::Error>
95    where
96        Self: Sized + Send + 'static,
97    {
98        UnsyncBoxBody::new(self)
99    }
100
101    /// Turn this body into [`Collected`] body which will collect all the DATA frames
102    /// and trailers.
103    fn collect(self) -> combinators::Collect<Self>
104    where
105        Self: Sized,
106    {
107        combinators::Collect {
108            body: self,
109            collected: Some(crate::Collected::default()),
110        }
111    }
112
113    /// Add trailers to the body.
114    ///
115    /// The trailers will be sent when all previous frames have been sent and the `trailers` future
116    /// resolves.
117    ///
118    /// # Example
119    ///
120    /// ```
121    /// use http::HeaderMap;
122    /// use http_body_util::{Full, BodyExt};
123    /// use bytes::Bytes;
124    ///
125    /// # #[tokio::main]
126    /// async fn main() {
127    /// let (tx, rx) = tokio::sync::oneshot::channel::<HeaderMap>();
128    ///
129    /// let body = Full::<Bytes>::from("Hello, World!")
130    ///     // add trailers via a future
131    ///     .with_trailers(async move {
132    ///         match rx.await {
133    ///             Ok(trailers) => Some(Ok(trailers)),
134    ///             Err(_err) => None,
135    ///         }
136    ///     });
137    ///
138    /// // compute the trailers in the background
139    /// tokio::spawn(async move {
140    ///     let _ = tx.send(compute_trailers().await);
141    /// });
142    ///
143    /// async fn compute_trailers() -> HeaderMap {
144    ///     // ...
145    ///     # unimplemented!()
146    /// }
147    /// # }
148    /// ```
149    fn with_trailers<F>(self, trailers: F) -> combinators::WithTrailers<Self, F>
150    where
151        Self: Sized,
152        F: std::future::Future<Output = Option<Result<http::HeaderMap, Self::Error>>>,
153    {
154        combinators::WithTrailers::new(self, trailers)
155    }
156
157    /// Turn this body into [`BodyStream`].
158    fn into_stream(self) -> BodyStream<Self>
159    where
160        Self: Sized,
161    {
162        BodyStream::new(self)
163    }
164
165    /// Turn this body into [`BodyDataStream`].
166    fn into_data_stream(self) -> BodyDataStream<Self>
167    where
168        Self: Sized,
169    {
170        BodyDataStream::new(self)
171    }
172
173    /// Creates a "fused" body.
174    ///
175    /// This [`Body`][http_body::Body] yields `Poll::Ready(None)` forever after the underlying
176    /// body yields `Poll::Ready(None)`, or an error `Poll::Ready(Some(Err(_)))`, once.
177    ///
178    /// See [`Fuse<B>`][combinators::Fuse] for more information.
179    fn fuse(self) -> combinators::Fuse<Self>
180    where
181        Self: Sized,
182    {
183        combinators::Fuse::new(self)
184    }
185}
186
187impl<T: ?Sized> BodyExt for T where T: http_body::Body {}
http_body_util/lib.rs

http_body_util/
lib.rs
