rws_tower/tower/
layer.rs

1//! [`TowerLayer`]: a Tower layer as rws middleware.
2
3use std::future::Future;
4use std::pin::Pin;
5
6use bytes::Bytes;
7use http_body_util::BodyExt;
8
9use super::{block_on, call, from_http_response, to_http_request, to_http_response, Body, BoxError};
10use crate::application::Application;
11use crate::middleware::Middleware;
12use crate::request::Request;
13use crate::response::Response;
14use crate::server::ConnectionInfo;
15
16/// A Tower [`Layer`](tower_layer::Layer) — tower-http's
17/// `SetResponseHeaderLayer`, `ValidateRequestHeaderLayer`, `TimeoutLayer`,
18/// `NormalizePathLayer`, … — as rws [`Middleware`]: `app.wrap(TowerLayer::new(layer))`.
19///
20/// Per request the layer wraps the rest of the stack ([`Next`]); the request
21/// goes in as `http::Request<Body>` and the response comes back through the
22/// layer as `http::Response<_>`. Streamed responses (files, SSE, proxied
23/// streams) stay streams. Request bodies reach the layer read into memory.
24/// It runs on a blocking-pool thread, never inline on the event loop.
25pub struct TowerLayer<L> {
26    layer: L,
27}
28
29impl<L> TowerLayer<L> {
30    pub fn new(layer: L) -> TowerLayer<L> {
31        TowerLayer { layer }
32    }
33}
34
35/// The rest of the rws stack, as the Tower service a layer wraps.
36pub struct Next<'a> {
37    app: &'a dyn Application,
38    connection: &'a ConnectionInfo,
39}
40
41impl<'a, ReqBody> tower_service::Service<http::Request<ReqBody>> for Next<'a>
42where
43    ReqBody: http_body::Body<Data = Bytes> + 'a,
44    ReqBody::Error: Into<BoxError>,
45{
46    type Response = http::Response<Body>;
47    type Error = BoxError;
48    type Future = Pin<Box<dyn Future<Output = Result<http::Response<Body>, BoxError>> + 'a>>;
49
50    fn poll_ready(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), BoxError>> {
51        std::task::Poll::Ready(Ok(()))
52    }
53
54    fn call(&mut self, request: http::Request<ReqBody>) -> Self::Future {
55        let (app, connection) = (self.app, self.connection);
56        Box::pin(async move {
57            let (parts, body) = request.into_parts();
58            let bytes = body.collect().await.map_err(Into::into)?.to_bytes();
59            let request = super::from_http_request(http::Request::from_parts(parts, Body::full(bytes))).await?;
60            let response = app.execute(&request, connection).map_err(BoxError::from)?;
61            Ok(to_http_response(response))
62        })
63    }
64}
65
66impl<L, ResBody> Middleware for TowerLayer<L>
67where
68    L: for<'a> tower_layer::Layer<Next<'a>> + Send + Sync,
69    for<'a> <L as tower_layer::Layer<Next<'a>>>::Service: tower_service::Service<http::Request<Body>, Response = http::Response<ResBody>>,
70    for<'a> <<L as tower_layer::Layer<Next<'a>>>::Service as tower_service::Service<http::Request<Body>>>::Error: Into<BoxError>,
71    ResBody: http_body::Body<Data = Bytes> + Send + 'static,
72    ResBody::Error: Into<BoxError>,
73{
74    fn handle(&self, request: &Request, connection: &ConnectionInfo, next: &dyn Application) -> Result<Response, String> {
75        let service = self.layer.layer(Next { app: next, connection });
76        let request = to_http_request(request.clone());
77        block_on(async move {
78            let response = call(service, request).await?;
79            from_http_response(response).await
80        })
81        .map_err(|e| format!("tower layer: {e}"))
82    }
83}
rws_tower/tower/layer.rs

rws_tower/tower/
layer.rs
