alux_http_hyper/
message.rs

1//! Reads a hyper request as the request a surface answers, and writes the answer back.
2
3use alux_http::{HttpMethod, HttpStatus};
4use alux_http_direct::{DirectBody, DirectError, DirectRequest, DirectResponse};
5use bytes::Bytes;
6use core::fmt::Display;
7use core::pin::pin;
8use futures::TryStreamExt;
9use http_body_util::combinators::UnsyncBoxBody;
10use http_body_util::{BodyExt, Full, StreamBody};
11use hyper::body::{Body, Frame};
12use hyper::header::{HeaderName, HeaderValue};
13use hyper::{Request, Response, StatusCode};
14use std::io::Error as IoError;
15
16/// The body a served answer carries, whether its bytes were in hand or are still to come.
17///
18/// A body still to come is whatever produces it, which one task reads in order, so what carries it
19/// need not be shared across threads.
20pub type HyperBody = UnsyncBoxBody<Bytes, IoError>;
21
22/// The answer this service produces, which is a hyper response like any other.
23pub type HyperAnswer = Response<HyperBody>;
24
25/// How many bytes of a request body are read before the request is refused.
26///
27/// A transport carrying no framework inherits no framework's policy, so it states one here rather
28/// than reading a body of any size into memory. A service that accepts more, or less, states what
29/// it accepts with [`HyperRoute::reading`](crate::HyperRoute::reading), which production is
30/// expected to do: this is a default, not a decision about a caller's uploads.
31pub(crate) const BODY_LIMIT: usize = 8 * 1024 * 1024;
32
33/// The status a body larger than what is read is answered with.
34const TOO_LARGE: HttpStatus = HttpStatus::new(413);
35
36/// Reads a hyper request as the request a surface answers.
37pub(crate) async fn asked<Sent>(request: Request<Sent>, reading: usize) -> Result<DirectRequest, DirectResponse>
38where
39    Sent: Body<Data = Bytes>,
40    Sent::Error: Display,
41{
42    let (head, sent) = request.into_parts();
43    let Some(method) = named(head.method.as_str()) else {
44        return Err(DirectError::method_not_allowed(head.uri.path()).into());
45    };
46    let mut asked = DirectRequest::new(method, head.uri.path());
47    if let Some(query) = head.uri.query() {
48        asked = asked.with_query(query);
49    }
50    for (name, value) in &head.headers {
51        if let Ok(value) = value.to_str() {
52            asked = asked.with_header(name.as_str(), value);
53        }
54    }
55
56    Ok(asked.with_body(read(sent, reading).await?))
57}
58
59/// Reads the body a caller sent, up to `reading` bytes, and refuses one larger than that.
60///
61/// Frames are read one at a time and counted as they arrive, so a body that never ends is refused
62/// at the byte the limit names rather than after it has been held whole.
63async fn read<Sent>(sent: Sent, reading: usize) -> Result<Vec<u8>, DirectResponse>
64where
65    Sent: Body<Data = Bytes>,
66    Sent::Error: Display,
67{
68    let mut body = Vec::new();
69    let mut sent = pin!(sent);
70    while let Some(frame) = sent.frame().await {
71        let frame = frame.map_err(|error| DirectResponse::from(DirectError::unreadable("body", &error.to_string())))?;
72        let Ok(data) = frame.into_data() else {
73            continue;
74        };
75        if body.len() + data.len() > reading {
76            let refused = format!("the body is larger than the {reading} bytes this service reads");
77
78            return Err(DirectError::new(TOO_LARGE, refused).into());
79        }
80        body.extend_from_slice(&data);
81    }
82
83    Ok(body)
84}
85
86/// Reads a method name as the method a program states, where it states one.
87fn named(method: &str) -> Option<HttpMethod> {
88    HttpMethod::ALL.iter().copied().find(|stated| stated.label() == method)
89}
90
91/// Writes the answer a surface produced as the response hyper sends.
92pub(crate) fn answered(answer: DirectResponse) -> HyperAnswer {
93    let status = StatusCode::from_u16(answer.status().code()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
94    let stated = answer.headers().map(|(name, value)| (name.to_owned(), value.to_owned())).collect::<Vec<_>>();
95    let mut response = Response::new(carried(answer.into_body()));
96    *response.status_mut() = status;
97    for (name, value) in stated {
98        if let (Ok(name), Ok(value)) = (HeaderName::from_bytes(name.as_bytes()), HeaderValue::from_str(&value)) {
99            response.headers_mut().append(name, value);
100        }
101    }
102
103    response
104}
105
106/// Writes what an answer carries as the body hyper sends, however the answer carries it.
107fn carried(body: DirectBody) -> HyperBody {
108    match body {
109        DirectBody::Stated(body) => Full::new(Bytes::from(body)).map_err(|never| match never {}).boxed_unsync(),
110        DirectBody::Produced(chunks) => {
111            StreamBody::new(chunks.map_ok(|chunk| Frame::data(Bytes::from(chunk)))).boxed_unsync()
112        }
113    }
114}
alux_http_hyper/message.rs

alux_http_hyper/
message.rs
