karyon_jsonrpc/server/http/
h1h2.rs

1//! HTTP/1.1 and HTTP/2 over TCP via hyper.
2
3use std::{net::SocketAddr, sync::Arc};
4
5use bytes::Bytes;
6use http_body_util::{BodyExt, Full, LengthLimitError, Limited};
7use hyper::{
8    body::{Body, Incoming},
9    service::service_fn,
10    Method, Request, Response, StatusCode,
11};
12use hyper_util::server::conn::auto::Builder as HttpConnBuilder;
13use log::{debug, error};
14
15#[cfg(feature = "tokio")]
16use hyper_util::rt::TokioTimer as HyperTimer;
17#[cfg(feature = "smol")]
18use smol_hyper::rt::SmolTimer as HyperTimer;
19
20use karyon_core::{
21    async_runtime::net::{TcpListener, TcpStream},
22    async_util::Backoff,
23};
24
25use crate::{
26    codec::{decode_body, JsonRpcCodec},
27    error::Result,
28    message,
29    server::{
30        http::{
31            encode_response, json_response, json_response_bytes, parse_error_response,
32            HyperExecutor, ERR_BODY_TOO_LARGE, ERR_ENCODE_RESPONSE, ERR_METHOD_NOT_ALLOWED,
33            ERR_READ_BODY, MAX_HTTP_BODY_SIZE,
34        },
35        Server,
36    },
37};
38
39pub(super) async fn accept_tcp<C: JsonRpcCodec>(
40    server: &Arc<Server>,
41    listener: &TcpListener,
42    codec: &C,
43    backoff: &Backoff,
44) {
45    match listener.accept().await {
46        Ok((stream, peer_addr)) => {
47            backoff.reset();
48            server
49                .task_group
50                .spawn(serve_task(server.clone(), stream, peer_addr, codec.clone()));
51        }
52        Err(err) => {
53            error!("Accept TCP connection: {err}");
54            backoff.sleep().await;
55        }
56    }
57}
58
59async fn serve_task<C: JsonRpcCodec>(
60    server: Arc<Server>,
61    stream: TcpStream,
62    peer_addr: SocketAddr,
63    codec: C,
64) -> Result<()> {
65    if let Err(err) = serve_conn(server, stream, peer_addr, codec).await {
66        error!("HTTP/1-2 from {peer_addr}: {err}");
67    }
68    Ok(())
69}
70
71async fn serve_conn<C: JsonRpcCodec>(
72    server: Arc<Server>,
73    stream: TcpStream,
74    peer_addr: SocketAddr,
75    codec: C,
76) -> Result<()> {
77    debug!("New HTTP/1-2 connection from {peer_addr}");
78
79    let task_group = server.task_group.clone();
80    let service = service_fn(move |req: Request<Incoming>| {
81        let server = server.clone();
82        let codec = codec.clone();
83        async move { handle_hyper_request(server, req, codec).await }
84    });
85
86    #[cfg(feature = "smol")]
87    let io = smol_hyper::rt::FuturesIo::new(stream);
88    #[cfg(feature = "tokio")]
89    let io = hyper_util::rt::TokioIo::new(stream);
90
91    let mut builder = HttpConnBuilder::new(HyperExecutor::new(task_group));
92    // hyper only applies its HTTP/1.1 header read timeout (30s) when a
93    // timer is set; without it a client can hold the connection by never
94    // finishing its headers.
95    builder.http1().timer(HyperTimer::new());
96    let conn = builder.serve_connection(io, service);
97
98    if let Err(err) = conn.await {
99        debug!("HTTP/1-2 from {peer_addr} closed: {err}");
100    }
101
102    Ok(())
103}
104
105async fn handle_hyper_request<C: JsonRpcCodec>(
106    server: Arc<Server>,
107    req: Request<Incoming>,
108    codec: C,
109) -> std::result::Result<Response<Full<Bytes>>, hyper::Error> {
110    if req.method() != Method::POST {
111        return Ok(json_response(
112            StatusCode::METHOD_NOT_ALLOWED,
113            ERR_METHOD_NOT_ALLOWED,
114        ));
115    }
116
117    // Fast reject when Content-Length already exceeds the cap.
118    if let Some(len) = req.body().size_hint().upper() {
119        if len > MAX_HTTP_BODY_SIZE {
120            return Ok(json_response(
121                StatusCode::PAYLOAD_TOO_LARGE,
122                ERR_BODY_TOO_LARGE,
123            ));
124        }
125    }
126
127    // Limited enforces the cap during read regardless of Content-Length.
128    let limited = Limited::new(req.into_body(), MAX_HTTP_BODY_SIZE as usize);
129    let body = match limited.collect().await {
130        Ok(collected) => collected.to_bytes(),
131        Err(err) => {
132            if err.downcast_ref::<LengthLimitError>().is_some() {
133                return Ok(json_response(
134                    StatusCode::PAYLOAD_TOO_LARGE,
135                    ERR_BODY_TOO_LARGE,
136                ));
137            }
138            return Ok(json_response(StatusCode::BAD_REQUEST, ERR_READ_BODY));
139        }
140    };
141
142    let response = match decode_body(&codec, &body) {
143        Ok(msg) => server.handle_request(None, msg).await,
144        Err(_) => parse_error_response(),
145    };
146    debug!("--> {response}");
147
148    let content_type = &server.config.http_config.content_type;
149    Ok(encode_hyper_response(&codec, content_type, &response))
150}
151
152fn encode_hyper_response<C: JsonRpcCodec>(
153    codec: &C,
154    content_type: &str,
155    resp: &message::Response,
156) -> Response<Full<Bytes>> {
157    match encode_response(codec, resp) {
158        Ok(body) => json_response_bytes(StatusCode::OK, content_type, body),
159        Err(err) => {
160            error!("Encode HTTP response: {err}");
161            json_response(StatusCode::INTERNAL_SERVER_ERROR, ERR_ENCODE_RESPONSE)
162        }
163    }
164}
karyon_jsonrpc/server/http/h1h2.rs

karyon_jsonrpc/server/http/
h1h2.rs
