hyper_body_utils/
lib.rs

1#![doc = include_str!("../README.md")]
2#![deny(missing_docs)]
3use bytes::Buf;
4use futures::ready;
5
6use bytes::Bytes;
7use futures::{stream, FutureExt, Stream, TryStreamExt};
8use std::pin::Pin;
9use std::task::{Context, Poll};
10
11#[cfg(feature = "http3")]
12use h3::client::RequestStream as ClientRequestStream;
13#[cfg(feature = "http3")]
14use h3::server::RequestStream as ServerRequestStream;
15#[cfg(feature = "http3")]
16use h3_quinn::RecvStream;
17
18pub use http_body_util::BodyExt;
19
20use http_body_util::StreamBody;
21use hyper::body::{Body, Frame, Incoming};
22
23use http_body_util::combinators::BoxBody;
24
25#[cfg(test)]
26mod tests;
27
28/// Enum to represent different types of HTTP bodies
29pub enum HttpBody {
30    /// Incoming body from hyper
31    Incoming(Incoming),
32    /// Boxed stream body
33    BoxedStream(BoxBody<Bytes, std::io::Error>),
34    /// QUIC client incoming stream
35    #[cfg(feature = "http3")]
36    QuicClientIncoming(ClientRequestStream<RecvStream, Bytes>),
37    /// QUIC server incoming stream
38    #[cfg(feature = "http3")]
39    QuicServerIncoming(ServerRequestStream<RecvStream, Bytes>),
40}
41
42impl HttpBody {
43    /// Create a new HttpBody from an Incoming body
44    pub fn from_incoming(incoming: Incoming) -> Self {
45        HttpBody::Incoming(incoming)
46    }
47
48    /// Create a new HttpBody from a QUIC client stream
49    #[cfg(feature = "http3")]
50    pub fn from_quic_client(stream: ClientRequestStream<RecvStream, Bytes>) -> Self {
51        HttpBody::QuicClientIncoming(stream)
52    }
53
54    /// Create a new HttpBody from a QUIC server stream
55    #[cfg(feature = "http3")]
56    pub fn from_quic_server(stream: ServerRequestStream<RecvStream, Bytes>) -> Self {
57        HttpBody::QuicServerIncoming(stream)
58    }
59
60    /// Create a new HttpBody from a text string
61    pub fn from_text(text: &str) -> Self {
62        Self::from_bytes(text.as_bytes())
63    }
64
65    /// Create a new empty HttpBody
66    pub fn empty() -> Self {
67        Self::from_bytes(&Bytes::new())
68    }
69
70    /// Create a new HttpBody from a stream
71    pub fn from_stream<S>(stream: S) -> Self
72    where
73        S: Stream<Item = Result<Frame<Bytes>, std::io::Error>> + Send + Sync + 'static,
74    {
75        let body = StreamBody::new(stream);
76        HttpBody::BoxedStream(BodyExt::boxed(body))
77    }
78
79    /// Create a new HttpBody from bytes
80    pub fn from_bytes(bytes: &[u8]) -> Self {
81        let all_bytes = Bytes::copy_from_slice(bytes);
82        let content = stream::iter(vec![Ok(all_bytes)]).map_ok(Frame::data);
83        let body = StreamBody::new(content);
84        HttpBody::BoxedStream(BodyExt::boxed(body))
85    }
86}
87
88impl Body for HttpBody {
89    type Data = Bytes;
90
91    type Error = std::io::Error;
92
93    fn poll_frame(
94        self: Pin<&mut Self>,
95        cx: &mut Context<'_>,
96    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
97        match self.get_mut() {
98            HttpBody::Incoming(incoming) => incoming
99                .frame()
100                .poll_unpin(cx)
101                .map_err(std::io::Error::other),
102
103            HttpBody::BoxedStream(stream) => {
104                stream.frame().poll_unpin(cx).map_err(std::io::Error::other)
105            },
106
107            #[cfg(feature = "http3")]
108            HttpBody::QuicClientIncoming(stream) => match ready!(stream.poll_recv_data(cx)) {
109                Ok(frame) => match frame {
110                    Some(mut frame) => Poll::Ready(Some(Ok(Frame::data(
111                        frame.copy_to_bytes(frame.remaining()),
112                    )))),
113                    None => {
114                        cx.waker().wake_by_ref();
115                        Poll::Ready(None)
116                    }
117                },
118                Err(e) => {
119                    println!("Error polling frame: {}", e);
120                    Poll::Ready(Some(Err(std::io::Error::other(e))))
121                }
122            },
123            
124            #[cfg(feature = "http3")]
125            HttpBody::QuicServerIncoming(stream) => match ready!(stream.poll_recv_data(cx)) {
126                Ok(frame) => match frame {
127                    Some(mut frame) => Poll::Ready(Some(Ok(Frame::data(
128                        frame.copy_to_bytes(frame.remaining()),
129                    )))),
130                    None => {
131                        cx.waker().wake_by_ref();
132                        Poll::Ready(None)
133                    }
134                },
135                Err(e) => Poll::Ready(Some(Err(std::io::Error::other(e)))),
136            },
137        }
138    }
139}
140
141impl Stream for HttpBody {
142    type Item = Result<Frame<Bytes>, std::io::Error>;
143
144    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
145        self.poll_frame(cx)
146    }
147}
hyper_body_utils/lib.rs

hyper_body_utils/
lib.rs
