rws_tower/tower/
service.rs

1//! [`TowerService`]: an rws app as a Tower service.
2
3use std::convert::Infallible;
4use std::future::Future;
5use std::pin::Pin;
6use std::sync::Arc;
7
8use bytes::Bytes;
9
10use super::{from_http_request, to_http_response, Body, BoxError};
11use crate::application::Application;
12use crate::response::Response;
13use crate::server::{Address, ConnectionInfo};
14
15/// An rws [`Application`] as a Tower `Service<http::Request<B>>` — to serve
16/// it with hyper, or nest it in an axum `Router` (`Router::fallback_service`).
17///
18/// The request body is read into memory first. The app runs as rws runs it:
19/// `execute_async` when it offers one, inline when it says so, otherwise on
20/// the blocking pool — so call it from inside a tokio runtime. An app error
21/// becomes a `500`; the service itself never fails. `ConnectionInfo` is
22/// [`TowerService::connection`]'s (default: unknown peer, port 0), since
23/// Tower carries no peer address of its own.
24pub struct TowerService<A> {
25    app: Arc<A>,
26    connection: ConnectionInfo,
27}
28
29impl<A> Clone for TowerService<A> {
30    fn clone(&self) -> Self {
31        TowerService { app: Arc::clone(&self.app), connection: self.connection.clone() }
32    }
33}
34
35impl<A> TowerService<A> {
36    pub fn new(app: A) -> TowerService<A> {
37        let unknown = Address { ip: "0.0.0.0".to_string(), port: 0 };
38        TowerService { app: Arc::new(app), connection: ConnectionInfo { client: unknown.clone(), server: unknown, request_size: 0, sni_hostname: None, tls: false } }
39    }
40
41    /// What handlers see as the connection.
42    pub fn connection(mut self, connection: ConnectionInfo) -> Self {
43        self.connection = connection;
44        self
45    }
46}
47
48fn status(code: u16, message: &str) -> http::Response<Body> {
49    let mut response = http::Response::new(Body::full(message.to_string()));
50    *response.status_mut() = http::StatusCode::from_u16(code).unwrap_or(http::StatusCode::INTERNAL_SERVER_ERROR);
51    response
52}
53
54impl<A, B> tower_service::Service<http::Request<B>> for TowerService<A>
55where
56    A: Application + Send + Sync + 'static,
57    B: http_body::Body<Data = Bytes> + Send + 'static,
58    B::Error: Into<BoxError>,
59{
60    type Response = http::Response<Body>;
61    type Error = Infallible;
62    type Future = Pin<Box<dyn Future<Output = Result<http::Response<Body>, Infallible>> + Send>>;
63
64    fn poll_ready(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Infallible>> {
65        std::task::Poll::Ready(Ok(()))
66    }
67
68    fn call(&mut self, request: http::Request<B>) -> Self::Future {
69        let (app, connection) = (Arc::clone(&self.app), self.connection.clone());
70        Box::pin(async move {
71            let request = match from_http_request(request).await {
72                Ok(request) => request,
73                Err(e) => return Ok(status(400, &format!("unreadable request body: {e}"))),
74            };
75            let offered = match app.execute_async(&request, &connection) {
76                Some(future) => Some(future.await),
77                None => None,
78            };
79            let result: Result<Response, String> = match offered {
80                Some(result) => result,
81                None if app.runs_inline(&request, &connection) => app.execute(&request, &connection),
82                None => tokio::task::spawn_blocking(move || app.execute(&request, &connection))
83                    .await
84                    .unwrap_or_else(|e| Err(format!("handler panicked: {e}"))),
85            };
86            Ok(match result {
87                Ok(response) => to_http_response(response),
88                Err(message) => status(500, &message),
89            })
90        })
91    }
92}
rws_tower/tower/service.rs

rws_tower/tower/
service.rs
