a3s_boot/adapters/
axum.rs

1use crate::{
2    BootApplication, BootError, BootRequest, BootResponse, HttpAdapter, HttpMethod, Result,
3    RouteDefinition,
4};
5use axum::body::{to_bytes, Body, HttpBody};
6use axum::extract::Request;
7use axum::http::{
8    header::{ALLOW, CONTENT_TYPE},
9    response::Builder as ResponseBuilder,
10    HeaderName, HeaderValue, Method, StatusCode,
11};
12use axum::response::Response;
13use axum::routing::{on, MethodFilter, MethodRouter};
14use axum::Router;
15use futures_util::StreamExt;
16use std::error::Error;
17use std::net::SocketAddr;
18
19/// Axum-backed HTTP adapter.
20///
21/// Axum is the default adapter, not the Boot kernel. Applications can replace
22/// it by implementing [`HttpAdapter`] for another backend.
23#[derive(Debug, Clone)]
24pub struct AxumAdapter {
25    body_limit: usize,
26}
27
28impl AxumAdapter {
29    pub fn new() -> Self {
30        Self {
31            body_limit: 1024 * 1024,
32        }
33    }
34
35    pub fn with_body_limit(mut self, body_limit: usize) -> Self {
36        self.body_limit = body_limit;
37        self
38    }
39}
40
41impl Default for AxumAdapter {
42    fn default() -> Self {
43        Self::new()
44    }
45}
46
47impl HttpAdapter for AxumAdapter {
48    type Output = Router;
49
50    fn build(&self, app: BootApplication) -> Result<Self::Output> {
51        let mut router = Router::new().fallback(not_found_fallback);
52        for route in app.routes().iter().cloned() {
53            let path = axum_route_path(route.path());
54            router = router.route(
55                &path,
56                route_to_method_router(route, app.clone(), self.body_limit),
57            );
58        }
59        let body_limit = self.body_limit;
60        let app = app.clone();
61        Ok(router.method_not_allowed_fallback(move |request: Request| {
62            let app = app.clone();
63            async move { dispatch_application(app, request, body_limit).await }
64        }))
65    }
66
67    fn serve(
68        &self,
69        app: BootApplication,
70        addr: SocketAddr,
71    ) -> crate::BoxFuture<'static, Result<()>> {
72        let adapter = self.clone();
73        Box::pin(async move {
74            let router = adapter.build(app)?;
75            let listener = tokio::net::TcpListener::bind(addr).await?;
76            axum::serve(listener, router).await?;
77            Ok(())
78        })
79    }
80}
81
82fn route_to_method_router(
83    route: RouteDefinition,
84    app: BootApplication,
85    body_limit: usize,
86) -> MethodRouter {
87    let method = route.method();
88    let dispatch = move |request: Request| {
89        let route = route.clone();
90        let app = app.clone();
91        async move { dispatch_route(route, app, request, body_limit).await }
92    };
93
94    on(method_filter(method), dispatch)
95}
96
97fn method_filter(method: HttpMethod) -> MethodFilter {
98    match method {
99        HttpMethod::Get => MethodFilter::GET,
100        HttpMethod::Post => MethodFilter::POST,
101        HttpMethod::Put => MethodFilter::PUT,
102        HttpMethod::Patch => MethodFilter::PATCH,
103        HttpMethod::Delete => MethodFilter::DELETE,
104        HttpMethod::Options => MethodFilter::OPTIONS,
105        HttpMethod::Head => MethodFilter::HEAD,
106    }
107}
108
109fn axum_route_path(path: &str) -> String {
110    let path = path.strip_prefix('/').unwrap_or(path);
111    if path.is_empty() {
112        return "/".to_string();
113    }
114
115    let segments = path
116        .split('/')
117        .enumerate()
118        .map(|(index, segment)| {
119            if segment.starts_with('{') && segment.ends_with('}') {
120                format!("{{p{index}}}")
121            } else {
122                segment.to_string()
123            }
124        })
125        .collect::<Vec<_>>();
126
127    format!("/{}", segments.join("/"))
128}
129
130async fn dispatch_route(
131    route: RouteDefinition,
132    app: BootApplication,
133    request: Request,
134    body_limit: usize,
135) -> Response {
136    let path = request.uri().path().to_string();
137    let is_head = request.method() == Method::HEAD;
138    let boot_request = match to_boot_request(request, body_limit).await {
139        Ok(request) => request,
140        Err(err) => return finalize_response(&app, &path, boot_error_response(err), is_head),
141    };
142
143    let response = match route.call(boot_request).await {
144        Ok(response) => to_axum_response(response),
145        Err(err) => boot_error_response(err),
146    };
147    finalize_response(&app, &path, response, is_head)
148}
149
150async fn dispatch_application(
151    app: BootApplication,
152    request: Request,
153    body_limit: usize,
154) -> Response {
155    let path = request.uri().path().to_string();
156    let is_head = request.method() == Method::HEAD;
157    let boot_request = match to_boot_request(request, body_limit).await {
158        Ok(request) => request,
159        Err(err) => return finalize_response(&app, &path, boot_error_response(err), is_head),
160    };
161
162    let response = to_axum_response(app.handle(boot_request).await);
163    finalize_response(&app, &path, response, is_head)
164}
165
166async fn not_found_fallback(request: Request) -> Response {
167    let is_head = request.method() == Method::HEAD;
168    let response = boot_error_response(BootError::NotFound(format!(
169        "{} {}",
170        request.method(),
171        request.uri().path()
172    )));
173    strip_head_body(is_head, response)
174}
175
176async fn to_boot_request(axum_request: Request, body_limit: usize) -> Result<BootRequest> {
177    let path = axum_request.uri().path().to_string();
178    let method =
179        HttpMethod::try_from(axum_request.method().clone()).map_err(|error| match error {
180            BootError::MethodNotAllowed(method) => {
181                BootError::MethodNotAllowed(format!("{method} {path}"))
182            }
183            error => error,
184        })?;
185    let query_string = axum_request.uri().query().map(str::to_string);
186    let mut headers = Vec::new();
187    for (name, value) in axum_request.headers() {
188        let value = value.to_str().map_err(|err| {
189            BootError::BadRequest(format!("invalid request header value for {name}: {err}"))
190        })?;
191        headers.push((name.as_str().to_string(), value.to_string()));
192    }
193    let mut boot_request = BootRequest::new(method, path);
194    if let Some(query_string) = query_string {
195        boot_request = boot_request.with_query_string(query_string);
196    }
197    for (name, value) in headers {
198        boot_request = if boot_request.header(&name).is_some() {
199            boot_request.append_header(name, value)
200        } else {
201            boot_request.with_header(name, value)
202        };
203    }
204
205    boot_request.validate_headers()?;
206    boot_request.validate_body_limit(body_limit)?;
207
208    let body = axum_request.into_body();
209    if body.size_hint().lower() > body_limit as u64 {
210        return Err(BootError::PayloadTooLarge(format!(
211            "request body exceeds {body_limit} bytes"
212        )));
213    }
214    let body = to_bytes(body, body_limit)
215        .await
216        .map_err(|err| map_body_error(err, body_limit))?
217        .to_vec();
218
219    let boot_request = boot_request.with_body(body);
220    boot_request.validate_with_body_limit(body_limit)?;
221    Ok(boot_request)
222}
223
224fn map_body_error(error: axum::Error, body_limit: usize) -> BootError {
225    if error
226        .source()
227        .is_some_and(|source| source.is::<http_body_util::LengthLimitError>())
228    {
229        BootError::PayloadTooLarge(format!("request body exceeds {body_limit} bytes"))
230    } else {
231        BootError::Adapter(error.to_string())
232    }
233}
234
235fn to_axum_response(response: BootResponse) -> Response {
236    if let Err(error) = response.validate() {
237        return error_response(StatusCode::INTERNAL_SERVER_ERROR, internal_message(error));
238    }
239    let is_streaming = response.is_streaming();
240    let status = match StatusCode::from_u16(response.status()) {
241        Ok(status) => status,
242        Err(err) => {
243            return error_response(
244                StatusCode::INTERNAL_SERVER_ERROR,
245                format!("invalid response status {}: {err}", response.status()),
246            );
247        }
248    };
249
250    let mut builder = Response::builder().status(status);
251    builder = match with_response_headers(builder, response.header_entries()) {
252        Ok(builder) => builder,
253        Err(message) => return error_response(StatusCode::INTERNAL_SERVER_ERROR, message),
254    };
255
256    let body = if is_streaming {
257        let Some(stream) = response.into_sse_stream() else {
258            return error_response(
259                StatusCode::INTERNAL_SERVER_ERROR,
260                "streaming response body has already been consumed".to_string(),
261            );
262        };
263        Body::from_stream(stream.map(|event| event.map(|event| event.encode())))
264    } else {
265        Body::from(response.into_body())
266    };
267
268    builder
269        .body(body)
270        .unwrap_or_else(|err| error_response(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))
271}
272
273fn with_response_headers<I, N, V>(
274    mut builder: ResponseBuilder,
275    headers: I,
276) -> std::result::Result<ResponseBuilder, String>
277where
278    I: IntoIterator<Item = (N, V)>,
279    N: AsRef<str>,
280    V: AsRef<str>,
281{
282    for (name, value) in headers {
283        let name = name.as_ref();
284        let value = value.as_ref();
285        let header_name = HeaderName::try_from(name)
286            .map_err(|err| format!("invalid response header name {name:?}: {err}"))?;
287        let header_value = HeaderValue::try_from(value)
288            .map_err(|err| format!("invalid response header value for {name:?}: {err}"))?;
289        builder = builder.header(header_name, header_value);
290    }
291
292    Ok(builder)
293}
294
295fn internal_message(error: BootError) -> String {
296    match error {
297        BootError::Internal(message) => message,
298        error => error.to_string(),
299    }
300}
301
302fn with_allow_header(app: &BootApplication, path: &str, mut response: Response) -> Response {
303    if response.status() != StatusCode::METHOD_NOT_ALLOWED {
304        return response;
305    }
306
307    let Some(allow) = app.allowed_methods_header(path) else {
308        return response;
309    };
310
311    match HeaderValue::try_from(allow) {
312        Ok(value) => {
313            response.headers_mut().insert(ALLOW, value);
314            response
315        }
316        Err(err) => error_response(
317            StatusCode::INTERNAL_SERVER_ERROR,
318            format!("invalid allow header value: {err}"),
319        ),
320    }
321}
322
323fn finalize_response(
324    app: &BootApplication,
325    path: &str,
326    response: Response,
327    is_head: bool,
328) -> Response {
329    strip_head_body(is_head, with_allow_header(app, path, response))
330}
331
332fn strip_head_body(is_head: bool, response: Response) -> Response {
333    if !is_head {
334        return response;
335    }
336
337    let (parts, _) = response.into_parts();
338    Response::from_parts(parts, Body::empty())
339}
340
341fn boot_error_response(error: BootError) -> Response {
342    to_axum_response(BootResponse::from_error(&error))
343}
344
345fn error_response(status: StatusCode, message: String) -> Response {
346    let mut response = Response::new(Body::from(message));
347    *response.status_mut() = status;
348    response.headers_mut().insert(
349        CONTENT_TYPE,
350        HeaderValue::from_static("text/plain; charset=utf-8"),
351    );
352    response
353}
354
355impl TryFrom<Method> for HttpMethod {
356    type Error = BootError;
357
358    fn try_from(method: Method) -> std::result::Result<Self, Self::Error> {
359        method.as_str().parse()
360    }
361}
a3s_boot/adapters/axum.rs

a3s_boot/adapters/
axum.rs
