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
15use karyon_core::async_runtime::net::{TcpListener, TcpStream};
16
17use crate::{
18    error::Result,
19    message,
20    server::{
21        http::{
22            json_response, json_response_bytes, HyperExecutor, ERR_BODY_TOO_LARGE,
23            ERR_METHOD_NOT_ALLOWED, ERR_READ_BODY, MAX_HTTP_BODY_SIZE,
24        },
25        Server, FAILED_TO_PARSE_ERROR_MSG,
26    },
27};
28
29pub(super) async fn accept_tcp(server: &Arc<Server>, listener: &TcpListener) {
30    match listener.accept().await {
31        Ok((stream, peer_addr)) => {
32            server
33                .task_group
34                .spawn(serve_task(server.clone(), stream, peer_addr));
35        }
36        Err(err) => {
37            error!("Accept TCP connection: {err}");
38        }
39    }
40}
41
42async fn serve_task(server: Arc<Server>, stream: TcpStream, peer_addr: SocketAddr) -> Result<()> {
43    if let Err(err) = serve_conn(server, stream, peer_addr).await {
44        error!("HTTP/1-2 from {peer_addr}: {err}");
45    }
46    Ok(())
47}
48
49async fn serve_conn(server: Arc<Server>, stream: TcpStream, peer_addr: SocketAddr) -> Result<()> {
50    debug!("New HTTP/1-2 connection from {peer_addr}");
51
52    let task_group = server.task_group.clone();
53    let service = service_fn(move |req: Request<Incoming>| {
54        let server = server.clone();
55        async move { handle_hyper_request(server, req).await }
56    });
57
58    #[cfg(feature = "smol")]
59    let io = smol_hyper::rt::FuturesIo::new(stream);
60    #[cfg(feature = "tokio")]
61    let io = hyper_util::rt::TokioIo::new(stream);
62
63    let builder = HttpConnBuilder::new(HyperExecutor::new(task_group));
64    let conn = builder.serve_connection(io, service);
65
66    if let Err(err) = conn.await {
67        debug!("HTTP/1-2 from {peer_addr} closed: {err}");
68    }
69
70    Ok(())
71}
72
73async fn handle_hyper_request(
74    server: Arc<Server>,
75    req: Request<Incoming>,
76) -> std::result::Result<Response<Full<Bytes>>, hyper::Error> {
77    if req.method() != Method::POST {
78        return Ok(json_response(
79            StatusCode::METHOD_NOT_ALLOWED,
80            ERR_METHOD_NOT_ALLOWED,
81        ));
82    }
83
84    // Fast reject when Content-Length already exceeds the cap.
85    if let Some(len) = req.body().size_hint().upper() {
86        if len > MAX_HTTP_BODY_SIZE {
87            return Ok(json_response(
88                StatusCode::PAYLOAD_TOO_LARGE,
89                ERR_BODY_TOO_LARGE,
90            ));
91        }
92    }
93
94    // Limited enforces the cap during read regardless of Content-Length.
95    let limited = Limited::new(req.into_body(), MAX_HTTP_BODY_SIZE as usize);
96    let body = match limited.collect().await {
97        Ok(collected) => collected.to_bytes(),
98        Err(err) => {
99            if err.downcast_ref::<LengthLimitError>().is_some() {
100                return Ok(json_response(
101                    StatusCode::PAYLOAD_TOO_LARGE,
102                    ERR_BODY_TOO_LARGE,
103                ));
104            }
105            return Ok(json_response(StatusCode::BAD_REQUEST, ERR_READ_BODY));
106        }
107    };
108
109    let msg: serde_json::Value = match serde_json::from_slice(&body) {
110        Ok(v) => v,
111        Err(_) => {
112            let resp = message::Response {
113                error: Some(message::Error {
114                    code: message::PARSE_ERROR_CODE,
115                    message: FAILED_TO_PARSE_ERROR_MSG.to_string(),
116                    data: None,
117                }),
118                ..Default::default()
119            };
120            let json = serde_json::to_vec(&resp).unwrap();
121            return Ok(json_response_bytes(StatusCode::OK, json));
122        }
123    };
124
125    let response = server.handle_request(None, msg).await;
126    debug!("--> {response}");
127
128    let json = serde_json::to_vec(&response).unwrap();
129    Ok(json_response_bytes(StatusCode::OK, json))
130}
karyon_jsonrpc/server/http/h1h2.rs

karyon_jsonrpc/server/http/
h1h2.rs
