hyper_body_utils/
lib.rs

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

hyper_body_utils/
lib.rs
