karyon_jsonrpc/client/http/
h2.rs

1//! HTTP/2 client over plain TCP (h2c with prior knowledge).
2//!
3//! One connection is opened when the client is built and shared by
4//! all requests as HTTP/2 streams. Opt in with `HttpConfig::http2`.
5
6use std::{net::SocketAddr, sync::Arc};
7
8use bytes::Bytes;
9use http_body_util::Full;
10use hyper::{
11    client::conn::http2::{handshake, Connection, SendRequest},
12    header::CONTENT_TYPE,
13    rt::{Read, Write},
14    Request,
15};
16use log::error;
17
18use karyon_core::{async_runtime::net::TcpStream, async_util::TaskGroup};
19use karyon_net::Endpoint;
20
21#[cfg(feature = "tokio")]
22use hyper_util::rt::TokioIo;
23#[cfg(feature = "smol")]
24use smol_hyper::rt::FuturesIo;
25
26use crate::{
27    client::http::{parse_response, HttpConfig},
28    codec::{encode_msg, JsonRpcCodec},
29    error::{Error, Result},
30    hyper_exec::HyperExecutor,
31    message,
32};
33
34pub(super) type H2SendRequest = SendRequest<Full<Bytes>>;
35
36/// Open the TCP connection and run the HTTP/2 handshake. The connection
37/// driver runs in `task_group`, so it stops with the client.
38pub(super) async fn connect(
39    endpoint: &Endpoint,
40    task_group: &Arc<TaskGroup>,
41) -> Result<H2SendRequest> {
42    let addr = SocketAddr::try_from(endpoint.clone())?;
43    let stream = TcpStream::connect(addr).await?;
44
45    #[cfg(feature = "smol")]
46    let io = FuturesIo::new(stream);
47    #[cfg(feature = "tokio")]
48    let io = TokioIo::new(stream);
49
50    let exec = HyperExecutor::new(task_group.clone());
51    let (sender, conn) = handshake(exec, io)
52        .await
53        .map_err(|e| Error::HttpError(e.to_string()))?;
54
55    task_group.spawn(driver_task(conn));
56    Ok(sender)
57}
58
59pub(super) async fn send<C: JsonRpcCodec>(
60    sender: &H2SendRequest,
61    endpoint: &Endpoint,
62    config: &HttpConfig,
63    msg: serde_json::Value,
64    codec: &C,
65) -> Result<message::Response> {
66    let body = encode_msg(codec, &msg)?;
67
68    let req = Request::post(endpoint.to_string())
69        .header(CONTENT_TYPE, config.content_type.as_str())
70        .body(Full::new(body))
71        .map_err(|e| Error::HttpError(e.to_string()))?;
72
73    // Each request is a new stream on the shared connection.
74    let response = sender
75        .clone()
76        .send_request(req)
77        .await
78        .map_err(|e| Error::HttpError(e.to_string()))?;
79
80    parse_response(response, codec).await
81}
82
83async fn driver_task<T>(conn: Connection<T, Full<Bytes>, HyperExecutor>)
84where
85    T: Read + Write + Unpin + Send + 'static,
86{
87    if let Err(err) = conn.await {
88        error!("HTTP/2 client connection error: {err}");
89    }
90}
karyon_jsonrpc/client/http/h2.rs

karyon_jsonrpc/client/http/
h2.rs
