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