rama_http_hyperium/into_rama/
body.rs

1//! Wrap an external [`http_body::Body`] so it implements rama's
2//! [`Body`](rama_body::Body), for consuming ecosystem bodies inside rama.
3
4use std::fmt;
5use std::pin::Pin;
6use std::task::{Context, Poll};
7
8use pin_project_lite::pin_project;
9use rama_http_types::body::http_body as rama_body;
10
11use super::TryIntoRamaHttp as _;
12
13fn size_hint_to_rama(hint: &http_body::SizeHint) -> rama_body::SizeHint {
14    let mut out = rama_body::SizeHint::new();
15    out.set_lower(hint.lower());
16    if let Some(upper) = hint.upper() {
17        out.set_upper(upper);
18    }
19    out
20}
21
22pin_project! {
23    /// Wraps an external [`http_body::Body`] so it implements rama's
24    /// [`Body`](rama_body::Body).
25    pub struct RamaBody<B> {
26        #[pin]
27        inner: B,
28    }
29}
30
31impl<B> RamaBody<B> {
32    /// Wrap an external `http_body` body.
33    pub const fn new(inner: B) -> Self {
34        Self { inner }
35    }
36
37    /// Unwrap back into the external body.
38    pub fn into_inner(self) -> B {
39        self.inner
40    }
41}
42
43impl<B: http_body::Body> rama_body::Body for RamaBody<B> {
44    type Data = B::Data;
45    type Error = RamaBodyError<B::Error>;
46
47    fn poll_frame(
48        self: Pin<&mut Self>,
49        cx: &mut Context<'_>,
50    ) -> Poll<Option<Result<rama_body::Frame<Self::Data>, Self::Error>>> {
51        match self.project().inner.poll_frame(cx) {
52            Poll::Ready(Some(Ok(frame))) => match frame.into_data() {
53                Ok(data) => Poll::Ready(Some(Ok(rama_body::Frame::data(data)))),
54                Err(frame) => match frame.into_trailers() {
55                    Ok(trailers) => Poll::Ready(Some(
56                        trailers
57                            .try_into_rama_http()
58                            .map(rama_body::Frame::trailers)
59                            .map_err(RamaBodyError::Trailers),
60                    )),
61                    // http frames are data or trailers; defensive for any future kind.
62                    Err(_) => Poll::Ready(None),
63                },
64            },
65            Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(RamaBodyError::Body(err)))),
66            Poll::Ready(None) => Poll::Ready(None),
67            Poll::Pending => Poll::Pending,
68        }
69    }
70
71    fn is_end_stream(&self) -> bool {
72        self.inner.is_end_stream()
73    }
74
75    fn size_hint(&self) -> rama_body::SizeHint {
76        size_hint_to_rama(&self.inner.size_hint())
77    }
78}
79
80/// Error produced by a [`RamaBody`]: either the wrapped external body failed, or
81/// its trailers couldn't be converted to the rama `HeaderMap`.
82#[derive(Debug)]
83pub enum RamaBodyError<E> {
84    /// The wrapped external body errored.
85    Body(E),
86    /// Trailer conversion to the rama `HeaderMap` failed.
87    Trailers(rama_http_types::Error),
88}
89
90impl<E: fmt::Display> fmt::Display for RamaBodyError<E> {
91    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
92        match self {
93            Self::Body(err) => write!(f, "http body error: {err}"),
94            Self::Trailers(err) => write!(f, "trailer conversion to rama failed: {err}"),
95        }
96    }
97}
98
99impl<E: std::error::Error + 'static> std::error::Error for RamaBodyError<E> {
100    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
101        match self {
102            Self::Body(err) => Some(err),
103            Self::Trailers(err) => Some(err),
104        }
105    }
106}
rama_http_hyperium/into_rama/body.rs

rama_http_hyperium/into_rama/
body.rs
