karyon_jsonrpc/client/http/
h1.rs

1//! HTTP/1.1 client.
2//!
3//! - smol: hyper's `client::conn::http1`. No connection pooling: each
4//!   request opens a fresh TCP connection.
5//! - tokio: `hyper_util::client::legacy::Client`, which handles DNS and
6//!   reuses connections. hyper's high-level Client needs a tokio
7//!   runtime, so smol cannot use it.
8
9#[cfg(feature = "smol")]
10use std::net::SocketAddr;
11
12use bytes::Bytes;
13use http_body_util::Full;
14use hyper::{
15    header::{CONTENT_TYPE, HOST},
16    Request,
17};
18
19#[cfg(feature = "smol")]
20use hyper::{
21    client::conn::http1::{handshake, Connection},
22    rt::{Read, Write},
23};
24#[cfg(feature = "tokio")]
25use hyper_util::client::legacy::{connect::HttpConnector, Client as HyperClient};
26#[cfg(feature = "smol")]
27use log::error;
28#[cfg(feature = "smol")]
29use smol_hyper::rt::FuturesIo;
30
31#[cfg(feature = "smol")]
32use karyon_core::{async_runtime::net::TcpStream, async_util::TaskGroup};
33use karyon_net::Endpoint;
34
35use crate::{
36    client::http::{parse_response, HttpConfig},
37    codec::{encode_msg, JsonRpcCodec},
38    error::{Error, Result},
39    message,
40};
41
42#[cfg(feature = "smol")]
43pub(super) async fn send<C: JsonRpcCodec>(
44    endpoint: &Endpoint,
45    config: &HttpConfig,
46    msg: serde_json::Value,
47    task_group: &TaskGroup,
48    codec: &C,
49) -> Result<message::Response> {
50    let req = build_request(endpoint, config, &msg, codec)?;
51
52    // Resolve the endpoint to a SocketAddr (handles both literal IPs
53    // and domain names via the system DNS).
54    let addr = SocketAddr::try_from(endpoint.clone())?;
55
56    let stream = TcpStream::connect(addr).await?;
57    let io = FuturesIo::new(stream);
58
59    let (mut sender, conn) = handshake(io)
60        .await
61        .map_err(|e| Error::HttpError(e.to_string()))?;
62
63    // hyper's HTTP/1.1 client splits I/O from request building: the
64    // returned `conn` future drives the wire protocol and must be
65    // polled concurrently, otherwise `sender` blocks forever. Spawn
66    // it via the Client task_group so it gets cancelled on stop.
67    task_group.spawn(driver_task(conn));
68
69    let response = sender
70        .send_request(req)
71        .await
72        .map_err(|e| Error::HttpError(e.to_string()))?;
73
74    parse_response(response, codec).await
75}
76
77#[cfg(feature = "tokio")]
78pub(super) async fn send<C: JsonRpcCodec>(
79    client: &HyperClient<HttpConnector, Full<Bytes>>,
80    endpoint: &Endpoint,
81    config: &HttpConfig,
82    msg: serde_json::Value,
83    codec: &C,
84) -> Result<message::Response> {
85    let req = build_request(endpoint, config, &msg, codec)?;
86
87    let response = client
88        .request(req)
89        .await
90        .map_err(|e| Error::HttpError(e.to_string()))?;
91
92    parse_response(response, codec).await
93}
94
95/// POST request with the encoded message as body.
96fn build_request<C: JsonRpcCodec>(
97    endpoint: &Endpoint,
98    config: &HttpConfig,
99    msg: &serde_json::Value,
100    codec: &C,
101) -> Result<Request<Full<Bytes>>> {
102    let body = encode_msg(codec, msg)?;
103
104    let mut req =
105        Request::post(endpoint.to_string()).header(CONTENT_TYPE, config.content_type.as_str());
106    if let Some(host) = &config.host {
107        req = req.header(HOST, host);
108    }
109    req.body(Full::new(body))
110        .map_err(|e| Error::HttpError(e.to_string()))
111}
112
113#[cfg(feature = "smol")]
114async fn driver_task<T>(conn: Connection<T, Full<Bytes>>)
115where
116    T: Read + Write + Unpin + Send + 'static,
117{
118    if let Err(err) = conn.await {
119        error!("HTTP client connection error: {err}");
120    }
121}
karyon_jsonrpc/client/http/h1.rs

karyon_jsonrpc/client/http/
h1.rs
