hyper_body_utils/
lib.rs

1#![doc = include_str!("../README.md")]
2#![deny(missing_docs)]
3#[cfg(any(feature = "compio-h3", feature = "generic-h3"))]
4use bytes::Buf as _;
5use bytes::Bytes;
6#[cfg(any(feature = "compio-h3", feature = "generic-h3"))]
7use futures::ready;
8#[cfg(feature = "compio")]
9use futures::StreamExt as _;
10use futures::{FutureExt, Stream};
11use http_body_util::Full;
12use hyper::body::{Body, Frame, Incoming};
13use std::{
14    io::Error,
15    pin::Pin,
16    task::{Context, Poll},
17};
18
19pub use http_body_util::BodyExt;
20
21#[cfg(test)]
22mod tests;
23
24/// Enum to represent different types of HTTP bodies
25pub enum HttpBody {
26    /// Standard body from hyper holding all request body in memory
27    Standard(Full<Bytes>),
28    /// Incoming body from hyper, mainly for client responses and server request bodies
29    Incoming(Incoming),
30    #[cfg(feature = "generic")]
31    /// Boxed stream body from pool based runtimes
32    GenericStream(http_body_util::combinators::BoxBody<Bytes, std::io::Error>),
33    #[cfg(feature = "compio")]
34    /// Bytes framed body from compio
35    CompioStream(futures::stream::BoxStream<'static, Result<Bytes, std::io::Error>>),
36    /// QUIC client incoming stream
37    #[cfg(feature = "generic-h3")]
38    GenericClient(h3::client::RequestStream<h3_quinn::RecvStream, Bytes>),
39    /// QUIC server incoming stream
40    #[cfg(feature = "generic-h3")]
41    GenericServer(h3::server::RequestStream<h3_quinn::RecvStream, Bytes>),
42    /// QUIC client incoming stream
43    #[cfg(feature = "compio-h3")]
44    CompioClient(compio_quic::h3::client::RequestStream<compio_quic::RecvStream, Bytes>),
45    /// QUIC server incoming stream
46    #[cfg(feature = "compio-h3")]
47    CompioServer(compio_quic::h3::server::RequestStream<compio_quic::RecvStream, Bytes>),
48}
49
50impl HttpBody {
51    /// Create a new HttpBody from an Incoming body, you can use this method
52    /// to hold client response body or server incoming request body.
53    ///
54    /// # Arguments
55    /// * `incoming` - The Incoming body to create the HttpBody from
56    ///
57    /// # Returns
58    /// * `HttpBody` - The created HttpBody
59    pub fn from_incoming(incoming: Incoming) -> Self {
60        HttpBody::Incoming(incoming)
61    }
62
63    /// Create a new HttpBody from a QUIC client stream, it is intended to be used
64    /// with pool based runtimes only.
65    ///
66    /// # Arguments
67    /// * `stream` - The QUIC client stream to create the HttpBody from
68    ///
69    /// # Returns
70    /// * `HttpBody` - The created HttpBody
71    #[cfg(feature = "generic-h3")]
72    pub fn from_generic_client(
73        stream: h3::client::RequestStream<h3_quinn::RecvStream, Bytes>,
74    ) -> Self {
75        HttpBody::GenericClient(stream)
76    }
77
78    /// Create a new HttpBody from a QUIC server stream, it is intended to be used
79    /// with pool based runtimes only.
80    ///
81    /// # Arguments
82    /// * `stream` - The QUIC server stream to create the HttpBody from
83    ///
84    /// # Returns
85    /// * `HttpBody` - The created HttpBody
86    #[cfg(feature = "generic-h3")]
87    pub fn from_generic_server(
88        stream: h3::server::RequestStream<h3_quinn::RecvStream, Bytes>,
89    ) -> Self {
90        HttpBody::GenericServer(stream)
91    }
92
93    /// Create a new HttpBody from a QUIC client stream, it is intended to be used
94    /// with Compio runtime only.
95    ///
96    /// # Arguments
97    /// * `stream` - The QUIC client stream to create the HttpBody from
98    ///
99    /// # Returns
100    /// * `HttpBody` - The created HttpBody
101    #[cfg(feature = "compio-h3")]
102    pub fn from_compio_client(
103        stream: compio_quic::h3::client::RequestStream<compio_quic::RecvStream, Bytes>,
104    ) -> Self {
105        HttpBody::CompioClient(stream)
106    }
107
108    /// Create a new HttpBody from a QUIC server stream, it is intended to be used
109    /// with Compio runtime only.
110    ///
111    /// # Arguments
112    /// * `stream` - The QUIC server stream to create the HttpBody from
113    ///
114    /// # Returns
115    /// * `HttpBody` - The created HttpBody
116    #[cfg(feature = "compio-h3")]
117    pub fn from_compio_server(
118        stream: compio_quic::h3::server::RequestStream<compio_quic::RecvStream, Bytes>,
119    ) -> Self {
120        HttpBody::CompioServer(stream)
121    }
122
123    /// Create a new HttpBody from a text string
124    ///
125    /// # Arguments
126    /// * `text` - The text to create the HttpBody from
127    ///
128    /// # Returns
129    /// * `HttpBody` - The created HttpBody
130    ///
131    /// # Notes
132    /// * This method is not intended for large amounts of data
133    pub fn from_text(text: &str) -> Self {
134        Self::from_bytes(text.as_bytes())
135    }
136
137    /// Create a new HttpBody from bytes
138    ///
139    /// # Arguments
140    /// * `bytes` - The bytes to create the HttpBody from
141    ///
142    /// # Returns
143    /// * `HttpBody` - The created HttpBody
144    ///
145    /// # Notes
146    /// * This method is not intended for large amounts of data
147    pub fn from_bytes(bytes: &[u8]) -> Self {
148        let all_bytes = Bytes::copy_from_slice(bytes);
149        HttpBody::Standard(Full::new(all_bytes))
150    }
151
152    #[cfg(feature = "generic")]
153    /// Create a new HttpBody from a stream
154    ///
155    /// # Arguments
156    /// * `stream` - The stream to create the HttpBody from
157    ///
158    /// # Returns
159    /// * `HttpBody` - The created HttpBody
160    ///
161    /// # Notes
162    /// * This method is intended for use with streams that are already boxed
163    /// * You can't clone the stream, so you can't use it multiple times
164    pub fn from_generic_stream<S>(stream: S) -> Self
165    where
166        S: Stream<Item = Result<Frame<Bytes>, Error>> + Send + Sync + 'static,
167    {
168        let body = http_body_util::StreamBody::new(stream);
169        HttpBody::GenericStream(body.boxed())
170    }
171
172    #[cfg(feature = "compio")]
173    /// Create a new HttpBody from a stream
174    ///
175    /// # Arguments
176    /// * `stream` - The stream to create the HttpBody from
177    ///
178    /// # Returns
179    /// * `HttpBody` - The created HttpBody
180    ///
181    /// # Notes
182    /// * This method is intended for use with streams that are already boxed
183    /// * You can't clone the stream, so you can't use it multiple times
184    pub fn from_compio_stream<S>(stream: S) -> Self
185    where
186        S: Stream<Item = Result<Bytes, Error>> + Send + 'static,
187    {
188        HttpBody::CompioStream(stream.boxed())
189    }
190
191    /// Create a new empty HttpBody
192    ///
193    /// # Returns
194    /// * `HttpBody` - The created empty HttpBody
195    pub fn empty() -> Self {
196        Self::from_bytes(&Bytes::new())
197    }
198
199    /// Try to clone the HttpBody, if it is a stream, it will return None
200    ///
201    /// # Returns
202    /// * `Option<HttpBody>` - Some(HttpBody) if it can be cloned,
203    pub fn try_clone(&self) -> Result<Self, Error> {
204        match self {
205            HttpBody::Standard(content) => Ok(HttpBody::Standard(content.clone())),
206            _ => Err(Error::new(
207                std::io::ErrorKind::Other,
208                "Cannot clone stream body",
209            )),
210        }
211    }
212}
213
214impl Body for HttpBody {
215    type Data = Bytes;
216
217    type Error = std::io::Error;
218
219    fn poll_frame(
220        self: Pin<&mut Self>,
221        cx: &mut Context<'_>,
222    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
223        match self.get_mut() {
224            HttpBody::Standard(full_body) => full_body.frame().poll_unpin(cx).map_err(Error::other),
225
226            HttpBody::Incoming(incoming) => incoming.frame().poll_unpin(cx).map_err(Error::other),
227
228            #[cfg(feature = "generic")]
229            HttpBody::GenericStream(stream) => stream.frame().poll_unpin(cx).map_err(Error::other),
230
231            #[cfg(feature = "compio")]
232            HttpBody::CompioStream(stream) => stream
233                .poll_next_unpin(cx)
234                .map(|b| b.map(|b| b.map(Frame::data))),
235
236            #[cfg(feature = "generic-h3")]
237            HttpBody::GenericClient(stream) => match ready!(stream.poll_recv_data(cx)) {
238                Ok(frame) => match frame {
239                    Some(mut frame) => Poll::Ready(Some(Ok(Frame::data(
240                        frame.copy_to_bytes(frame.remaining()),
241                    )))),
242                    None => {
243                        cx.waker().wake_by_ref();
244                        Poll::Ready(None)
245                    }
246                },
247                Err(e) => Poll::Ready(Some(Err(Error::other(e)))),
248            },
249
250            #[cfg(feature = "generic-h3")]
251            HttpBody::GenericServer(stream) => match ready!(stream.poll_recv_data(cx)) {
252                Ok(frame) => match frame {
253                    Some(mut frame) => Poll::Ready(Some(Ok(Frame::data(
254                        frame.copy_to_bytes(frame.remaining()),
255                    )))),
256                    None => {
257                        cx.waker().wake_by_ref();
258                        Poll::Ready(None)
259                    }
260                },
261                Err(e) => Poll::Ready(Some(Err(Error::other(e)))),
262            },
263
264            #[cfg(feature = "compio-h3")]
265            HttpBody::CompioClient(stream) => match ready!(stream.poll_recv_data(cx)) {
266                Ok(frame) => match frame {
267                    Some(mut frame) => Poll::Ready(Some(Ok(Frame::data(
268                        frame.copy_to_bytes(frame.remaining()),
269                    )))),
270                    None => {
271                        cx.waker().wake_by_ref();
272                        Poll::Ready(None)
273                    }
274                },
275                Err(e) => Poll::Ready(Some(Err(Error::other(e)))),
276            },
277
278            #[cfg(feature = "compio-h3")]
279            HttpBody::CompioServer(stream) => match ready!(stream.poll_recv_data(cx)) {
280                Ok(frame) => match frame {
281                    Some(mut frame) => Poll::Ready(Some(Ok(Frame::data(
282                        frame.copy_to_bytes(frame.remaining()),
283                    )))),
284                    None => {
285                        cx.waker().wake_by_ref();
286                        Poll::Ready(None)
287                    }
288                },
289                Err(e) => Poll::Ready(Some(Err(Error::other(e)))),
290            },
291        }
292    }
293}
294
295impl Stream for HttpBody {
296    type Item = Result<Frame<Bytes>, Error>;
297
298    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
299        self.poll_frame(cx)
300    }
301}
302
303impl From<&str> for HttpBody {
304    fn from(value: &str) -> Self {
305        HttpBody::from_text(value)
306    }
307}
308
309impl From<String> for HttpBody {
310    fn from(value: String) -> Self {
311        HttpBody::from_text(&value)
312    }
313}
314
315impl From<&[u8]> for HttpBody {
316    fn from(value: &[u8]) -> Self {
317        HttpBody::from_bytes(value)
318    }
319}
320
321impl From<Vec<u8>> for HttpBody {
322    fn from(value: Vec<u8>) -> Self {
323        HttpBody::from_bytes(&value)
324    }
325}
326
327impl From<Bytes> for HttpBody {
328    fn from(value: Bytes) -> Self {
329        HttpBody::from_bytes(&value)
330    }
331}
332
333#[cfg(feature = "compio")]
334impl From<compio::fs::File> for HttpBody {
335    fn from(value: compio::fs::File) -> Self {
336        let stream = from_asyncread(value);
337        HttpBody::from_compio_stream(send_wrapper::SendWrapper::new(stream))
338    }
339}
340
341#[cfg(feature = "compio")]
342fn from_asyncread<R>(reader: R) -> impl Stream<Item = Result<Bytes, Error>>
343where
344    R: compio::io::AsyncReadAt,
345{
346    async_fn_stream::try_fn_stream(|emitter| async move {
347        let mut pos = 0;
348        loop {
349            let buf = Vec::with_capacity(4096);
350            let compio::BufResult(res, buffer) = reader.read_at(buf, pos).await;
351            let len = res?;
352            if len == 0 {
353                break Ok(());
354            }
355            pos += len as u64;
356            emitter.emit(Bytes::from(buffer)).await
357        }
358    })
359}
hyper_body_utils/lib.rs

hyper_body_utils/
lib.rs
