product_os_http_body/
body_bytes.rs

1//! Concrete implementation of HTTP body types.
2//!
3//! This module provides [`BodyBytes`], a flexible HTTP body type that can represent
4//! various kinds of bodies (empty, full, streaming, etc.) with a unified interface.
5
6// Adapted from hudsucker
7
8use crate::error::BodyError as Error;
9use alloc::borrow::ToOwned;
10use alloc::string::String;
11use bytes::Bytes;
12use http_body::{Body, Frame, SizeHint};
13use http_body_util::{BodyExt, Collected, Empty, Full};
14
15#[cfg(feature = "stream")]
16use alloc::boxed::Box;
17#[cfg(any(feature = "stream", feature = "incoming"))]
18use alloc::string::ToString;
19#[cfg(any(feature = "stream", feature = "box_body"))]
20use futures::Stream;
21#[cfg(feature = "stream")]
22use http_body_util::BodyStream;
23#[cfg(any(feature = "stream", feature = "box_body"))]
24use http_body_util::StreamBody;
25
26use product_os_http::{Request, Response};
27
28#[cfg(feature = "stream")]
29use crate::data_stream::{DataStream, DataStreamBody};
30#[cfg(any(feature = "incoming", feature = "stream"))]
31use hyper::body::Incoming;
32
33#[cfg(feature = "box_body")]
34use futures::TryStreamExt;
35#[cfg(feature = "box_body")]
36use futures_util::TryStream;
37#[cfg(feature = "box_body")]
38use http_body_util::combinators::BoxBody;
39
40#[cfg(feature = "std")]
41use std::{pin::Pin, task::Poll};
42
43#[derive(Debug)]
44enum Internal {
45    Collected(Collected<Bytes>),
46    Empty(Empty<Bytes>),
47    Full(Full<Bytes>),
48    String(String),
49    #[cfg(feature = "stream")]
50    BodyStream(BodyStream<Incoming>),
51    #[cfg(feature = "stream")]
52    Stream(StreamBody<DataStream<DataStreamBody>>),
53    #[cfg(feature = "box_body")]
54    BoxBody(BoxBody<Bytes, Error>),
55    #[cfg(feature = "incoming")]
56    Incoming(Incoming),
57}
58
59/// A concrete implementation of an HTTP body.
60///
61/// `BodyBytes` is an enum-backed type that can represent various kinds of HTTP bodies:
62/// - Empty bodies
63/// - Full bodies with all data available upfront
64/// - Streaming bodies from various sources
65/// - Collected bodies
66/// - Boxed bodies for type erasure
67///
68/// # Examples
69///
70/// ## Creating an empty body
71///
72/// ```rust
73/// use product_os_http_body::BodyBytes;
74///
75/// let body = BodyBytes::empty();
76/// ```
77///
78/// ## Creating a body from bytes
79///
80/// ```rust
81/// use product_os_http_body::{BodyBytes, Bytes};
82///
83/// let body = BodyBytes::new(Bytes::from("Hello, World!"));
84/// ```
85///
86/// ## Creating a body from a string
87///
88/// ```rust
89/// use product_os_http_body::BodyBytes;
90///
91/// let body = BodyBytes::from("string data");
92/// let body2 = BodyBytes::from(String::from("owned string"));
93/// ```
94///
95/// # Feature-gated functionality
96///
97/// Some constructors and conversions are only available with specific features:
98///
99/// - `stream`: Enables `new_stream()` and `from_stream()` for streaming bodies
100/// - `box_body`: Enables conversions from `BoxBody`
101/// - `incoming`: Enables conversions from `hyper::body::Incoming`
102///
103/// # Body trait implementation
104///
105/// `BodyBytes` implements the `http_body::Body` trait (when the `std` feature is enabled),
106/// allowing it to be used anywhere an HTTP body is expected.
107#[derive(Debug)]
108pub struct BodyBytes {
109    inner: Internal,
110}
111
112impl Default for BodyBytes {
113    fn default() -> Self {
114        Self::empty()
115    }
116}
117
118impl BodyBytes {
119    pub fn empty() -> Self {
120        Self::from(Empty::new())
121    }
122
123    pub fn new(bytes: Bytes) -> Self {
124        Self {
125            inner: Internal::Full(Full::new(bytes)),
126        }
127    }
128
129    #[cfg(feature = "stream")]
130    pub fn new_stream<
131        E: Into<Box<dyn core::error::Error + core::marker::Send + Sync + 'static>>,
132    >(
133        s: impl Stream<Item = Result<Bytes, E>> + Send + 'static,
134    ) -> Self {
135        let stream_body = DataStreamBody::stream(s);
136        let data_stream = DataStream(stream_body);
137
138        Self {
139            inner: Internal::Stream(StreamBody::new(data_stream)),
140        }
141    }
142
143    pub async fn convert_to_bytes(self) -> Result<Bytes, Error> {
144        match self.inner {
145            Internal::String(s) => Ok(Bytes::from(s.to_owned())),
146            Internal::Empty(e) => {
147                let bytes = e.collect().await.unwrap().to_bytes();
148                Ok(bytes)
149            }
150            Internal::Full(f) => {
151                let bytes = f.collect().await.unwrap().to_bytes();
152                Ok(bytes)
153            }
154            Internal::Collected(c) => {
155                let bytes = c.collect().await.unwrap().to_bytes();
156                Ok(bytes)
157            }
158            #[cfg(feature = "stream")]
159            Internal::BodyStream(b) => {
160                let bytes = BodyExt::collect(b).await.unwrap().to_bytes();
161                Ok(bytes)
162            }
163            #[cfg(feature = "stream")]
164            Internal::Stream(b) => {
165                let bytes = BodyExt::collect(b).await.unwrap().to_bytes();
166                Ok(bytes)
167            }
168            #[cfg(feature = "box_body")]
169            Internal::BoxBody(b) => {
170                let bytes = b.collect().await.unwrap().to_bytes();
171                Ok(bytes)
172            }
173            #[cfg(feature = "incoming")]
174            Internal::Incoming(i) => {
175                let bytes = i.collect().await.unwrap().to_bytes();
176                Ok(bytes)
177            }
178        }
179    }
180
181    pub async fn as_bytes(&mut self) -> Result<Bytes, Error> {
182        match &mut self.inner {
183            Internal::String(s) => Ok(Bytes::from(s.to_owned())),
184            Internal::Empty(e) => {
185                let bytes = e.collect().await.unwrap().to_bytes();
186                self.inner = Internal::Empty(Empty::new());
187                Ok(bytes)
188            }
189            Internal::Full(f) => {
190                let bytes = f.collect().await.unwrap().to_bytes();
191                self.inner = Internal::Full(Full::new(bytes.to_owned()));
192                Ok(bytes)
193            }
194            Internal::Collected(c) => {
195                let bytes = c.collect().await.unwrap().to_bytes();
196                self.inner =
197                    Internal::Collected(Full::new(bytes.to_owned()).collect().await.unwrap());
198                Ok(bytes)
199            }
200            #[cfg(feature = "stream")]
201            Internal::BodyStream(b) => {
202                let bytes = BodyExt::collect(b).await.unwrap().to_bytes();
203                self.inner =
204                    Internal::Collected(Full::new(bytes.to_owned()).collect().await.unwrap());
205                Ok(bytes)
206            }
207            #[cfg(feature = "stream")]
208            Internal::Stream(b) => {
209                let bytes = BodyExt::collect(b).await.unwrap().to_bytes();
210                self.inner =
211                    Internal::Collected(Full::new(bytes.to_owned()).collect().await.unwrap());
212                Ok(bytes)
213            }
214            #[cfg(feature = "box_body")]
215            Internal::BoxBody(_b) => {
216                /*
217                let bytes = b.collect().await.unwrap().to_bytes();
218                self.inner = Internal::BoxBody(BoxBody::new(Full::new(bytes.to_owned())));
219                bytes
220                */
221                Err(Error::DecodeToBytes(String::from(
222                    "Box body not yet implemented",
223                )))
224            }
225            #[cfg(feature = "incoming")]
226            Internal::Incoming(_i) => Err(Error::DecodeToBytes(String::from(
227                "Incoming not yet implemented",
228            ))),
229        }
230    }
231
232    #[cfg(feature = "box_body")]
233    pub fn from_stream<S>(stream: S) -> Self
234    where
235        S: TryStream + Send + Sync + 'static,
236        S::Ok: Into<Bytes>,
237        S::Error: Into<Error>,
238    {
239        Self {
240            inner: Internal::BoxBody(BoxBody::new(StreamBody::new(
241                stream
242                    .map_ok(Into::into)
243                    .map_ok(Frame::data)
244                    .map_err(Into::into),
245            ))),
246        }
247    }
248}
249
250impl Body for BodyBytes {
251    type Data = Bytes;
252    type Error = Error;
253
254    fn poll_frame(
255        mut self: Pin<&mut Self>,
256        cx: &mut std::task::Context<'_>,
257    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
258        match &mut self.inner {
259            Internal::Collected(body) => Pin::new(body).poll_frame(cx).map_err(|e| match e {}),
260            Internal::Empty(body) => Pin::new(body).poll_frame(cx).map_err(|e| match e {}),
261            Internal::Full(body) => Pin::new(body).poll_frame(cx).map_err(|e| match e {}),
262            Internal::String(body) => Pin::new(body).poll_frame(cx).map_err(|e| match e {}),
263            #[cfg(feature = "stream")]
264            Internal::BodyStream(body) => match Pin::new(body).poll_frame(cx) {
265                Poll::Ready(Some(ready)) => match ready {
266                    Ok(frame) => Poll::Ready(Some(Ok(frame))),
267                    Err(e) => Poll::Ready(Some(Err(Error::Network(e.to_string())))),
268                },
269                Poll::Ready(None) => Poll::Ready(None),
270                Poll::Pending => Poll::Pending,
271            },
272            #[cfg(feature = "stream")]
273            Internal::Stream(body) => match Pin::new(body).poll_frame(cx) {
274                Poll::Ready(Some(ready)) => match ready {
275                    Ok(frame) => Poll::Ready(Some(Ok(frame))),
276                    Err(e) => Poll::Ready(Some(Err(Error::Network(e.to_string())))),
277                },
278                Poll::Ready(None) => Poll::Ready(None),
279                Poll::Pending => Poll::Pending,
280            },
281            #[cfg(feature = "box_body")]
282            Internal::BoxBody(body) => Pin::new(body).poll_frame(cx),
283            #[cfg(feature = "incoming")]
284            Internal::Incoming(body) => Pin::new(body)
285                .poll_frame(cx)
286                .map_err(|e| Error::Network(e.to_string())),
287        }
288    }
289
290    fn is_end_stream(&self) -> bool {
291        match &self.inner {
292            Internal::Collected(body) => body.is_end_stream(),
293            Internal::Empty(body) => body.is_end_stream(),
294            Internal::Full(body) => body.is_end_stream(),
295            Internal::String(body) => body.is_end_stream(),
296            #[cfg(feature = "stream")]
297            Internal::BodyStream(body) => body.is_end_stream(),
298            #[cfg(feature = "stream")]
299            Internal::Stream(body) => body.is_end_stream(),
300            #[cfg(feature = "box_body")]
301            Internal::BoxBody(body) => body.is_end_stream(),
302            #[cfg(feature = "incoming")]
303            Internal::Incoming(body) => body.is_end_stream(),
304        }
305    }
306
307    fn size_hint(&self) -> SizeHint {
308        match &self.inner {
309            Internal::Collected(body) => body.size_hint(),
310            Internal::Empty(body) => body.size_hint(),
311            Internal::Full(body) => body.size_hint(),
312            Internal::String(body) => body.size_hint(),
313            #[cfg(feature = "stream")]
314            Internal::BodyStream(body) => Body::size_hint(body),
315            #[cfg(feature = "stream")]
316            Internal::Stream(body) => Body::size_hint(body),
317            #[cfg(feature = "box_body")]
318            Internal::BoxBody(body) => body.size_hint(),
319            #[cfg(feature = "incoming")]
320            Internal::Incoming(body) => body.size_hint(),
321        }
322    }
323}
324
325impl From<Collected<Bytes>> for BodyBytes {
326    fn from(value: Collected<Bytes>) -> Self {
327        Self {
328            inner: Internal::Collected(value),
329        }
330    }
331}
332
333impl From<Empty<Bytes>> for BodyBytes {
334    fn from(value: Empty<Bytes>) -> Self {
335        Self {
336            inner: Internal::Empty(value),
337        }
338    }
339}
340
341impl From<Full<Bytes>> for BodyBytes {
342    fn from(value: Full<Bytes>) -> Self {
343        Self {
344            inner: Internal::Full(value),
345        }
346    }
347}
348
349#[cfg(feature = "stream")]
350impl From<BodyStream<Incoming>> for BodyBytes {
351    fn from(value: BodyStream<Incoming>) -> Self {
352        Self {
353            inner: Internal::BodyStream(value),
354        }
355    }
356}
357
358#[cfg(feature = "stream")]
359impl From<DataStreamBody> for BodyBytes {
360    fn from(value: DataStreamBody) -> Self {
361        let data_stream = DataStream(value);
362
363        Self {
364            inner: Internal::Stream(StreamBody::new(data_stream)),
365        }
366    }
367}
368
369#[cfg(feature = "box_body")]
370impl From<BoxBody<Bytes, Error>> for BodyBytes {
371    fn from(value: BoxBody<Bytes, Error>) -> Self {
372        Self {
373            inner: Internal::BoxBody(value),
374        }
375    }
376}
377
378#[cfg(feature = "incoming")]
379impl From<Incoming> for BodyBytes {
380    fn from(value: Incoming) -> Self {
381        Self {
382            inner: Internal::Incoming(value),
383        }
384    }
385}
386
387impl From<Bytes> for BodyBytes {
388    fn from(value: Bytes) -> Self {
389        Self {
390            inner: Internal::Full(Full::new(value)),
391        }
392    }
393}
394
395impl From<alloc::vec::Vec<u8>> for BodyBytes {
396    fn from(value: alloc::vec::Vec<u8>) -> Self {
397        Self {
398            inner: Internal::Full(Full::new(Bytes::from(value))),
399        }
400    }
401}
402
403impl From<String> for BodyBytes {
404    fn from(value: String) -> Self {
405        Self {
406            inner: Internal::String(value),
407        }
408    }
409}
410
411impl From<&'static str> for BodyBytes {
412    fn from(value: &'static str) -> Self {
413        Self {
414            inner: Internal::Full(Full::new(Bytes::from_static(value.as_bytes()))),
415        }
416    }
417}
418
419impl From<&'static [u8]> for BodyBytes {
420    fn from(value: &'static [u8]) -> Self {
421        Self {
422            inner: Internal::Full(Full::new(Bytes::from_static(value))),
423        }
424    }
425}
426
427impl<T> From<Request<T>> for BodyBytes
428where
429    T: Into<BodyBytes>,
430{
431    fn from(value: Request<T>) -> Self {
432        value.into_body().into()
433    }
434}
435
436impl<T> From<Response<T>> for BodyBytes
437where
438    T: Into<BodyBytes>,
439{
440    fn from(value: Response<T>) -> Self {
441        value.into_body().into()
442    }
443}
444
445#[cfg(feature = "box_body")]
446impl<S> From<StreamBody<S>> for BodyBytes
447where
448    S: Stream<Item = Result<Frame<Bytes>, Error>> + Send + Sync + 'static,
449{
450    fn from(value: StreamBody<S>) -> Self {
451        Self {
452            inner: Internal::BoxBody(BoxBody::new(value)),
453        }
454    }
455}
456
457impl Clone for BodyBytes {
458    fn clone(&self) -> Self {
459        match &self.inner {
460            /*
461            Internal::Collected(body) => {
462                Self {
463                    inner: Internal::Collected(body.to_owned())
464                }
465            }
466            */
467            Internal::Empty(body) => Self {
468                inner: Internal::Empty(body.to_owned()),
469            },
470            Internal::Full(body) => Self {
471                inner: Internal::Full(body.to_owned()),
472            },
473            Internal::String(body) => Self {
474                inner: Internal::String(body.to_owned()),
475            },
476            #[cfg(feature = "box_body")]
477            Internal::BoxBody(_body) => {
478                panic!("Clone not yet implemented for BoxBody")
479            }
480            #[cfg(feature = "incoming")]
481            Internal::Incoming(_body) => {
482                panic!("Clone not yet implemented for Incoming")
483            }
484            _ => panic!("Unable to handle clone"),
485        }
486    }
487}
488
489#[cfg(feature = "axum")]
490impl axum_core::response::IntoResponse for BodyBytes {
491    fn into_response(self) -> axum_core::response::Response {
492        axum_core::response::Response::new(axum_core::body::Body::new(self))
493    }
494}
product_os_http_body/body_bytes.rs

product_os_http_body/
body_bytes.rs
