rama_http/layer/
collect_body.rs

1//! Collect the http `Body`
2
3use crate::{Body, Request, Response, StreamingBody, body::util::BodyExt};
4use rama_core::{
5    Layer, Service,
6    error::{BoxError, ErrorContext},
7};
8use rama_utils::macros::define_inner_service_accessors;
9
10/// An http layer to collect the http `Body`
11#[derive(Debug, Clone, Default)]
12#[non_exhaustive]
13pub struct CollectBodyLayer;
14
15impl CollectBodyLayer {
16    /// Create a new [`CollectBodyLayer`].
17    #[must_use]
18    pub const fn new() -> Self {
19        Self
20    }
21}
22
23impl<S> Layer<S> for CollectBodyLayer {
24    type Service = CollectBody<S>;
25
26    fn layer(&self, inner: S) -> Self::Service {
27        CollectBody::new(inner)
28    }
29}
30
31/// Service to collect the http `Body`
32#[derive(Debug, Clone)]
33pub struct CollectBody<S> {
34    inner: S,
35}
36
37impl<S> CollectBody<S> {
38    /// Create a new [`CollectBody`].
39    pub const fn new(service: S) -> Self {
40        Self { inner: service }
41    }
42
43    define_inner_service_accessors!();
44}
45
46impl<S, ReqBody, ResBody> Service<Request<ReqBody>> for CollectBody<S>
47where
48    S: Service<Request<ReqBody>, Output = Response<ResBody>, Error: Into<BoxError>>,
49    ReqBody: Send + 'static,
50    ResBody: StreamingBody<Data: Send, Error: std::error::Error + Send + Sync + 'static>
51        + Send
52        + Sync
53        + 'static,
54{
55    type Output = Response;
56    type Error = BoxError;
57
58    async fn serve(&self, req: Request<ReqBody>) -> Result<Self::Output, Self::Error> {
59        let resp = self
60            .inner
61            .serve(req)
62            .await
63            .context("CollectBody::inner:serve")?;
64        let (parts, body) = resp.into_parts();
65        let bytes = body.collect().await.context("collect body")?.to_bytes();
66        let body = Body::from(bytes);
67        Ok(Response::from_parts(parts, body))
68    }
69}
rama_http/layer/collect_body.rs

rama_http/layer/
collect_body.rs
