karyon_jsonrpc/client/http/
h3.rs

1//! HTTP/3 client over QUIC via h3 / h3-quinn.
2
3use std::net::SocketAddr;
4use std::sync::Arc;
5
6use async_channel::Receiver;
7use bytes::Bytes;
8use h3::client::{RequestStream, SendRequest};
9use h3_quinn::{BidiStream, SendStream};
10use hyper::{header::CONTENT_TYPE, Request, StatusCode};
11use log::debug;
12use serde_json::Value;
13
14use karyon_core::async_util::{select, Either, TaskGroup};
15use karyon_net::{ByteBuffer, Endpoint};
16
17use crate::{
18    client::http::HttpConfig,
19    client::subscriptions::{decode_notification, NotificationDecoder, Subscription},
20    codec::{encode_msg, extend_buf, JsonRpcCodec},
21    error::{Error, Result},
22    message,
23};
24
25pub(super) type H3SendRequest = SendRequest<h3_quinn::OpenStreams, Bytes>;
26pub(super) type H3RecvStream = RequestStream<h3_quinn::RecvStream, Bytes>;
27pub(super) type H3SendStream = RequestStream<SendStream<Bytes>, Bytes>;
28type H3Stream = RequestStream<BidiStream<Bytes>, Bytes>;
29
30/// Connect to an HTTP/3 server and return a request sender. A
31/// background task drives the connection until close.
32pub(super) async fn connect(
33    endpoint: &Endpoint,
34    quic_config: &karyon_net::quic::ClientQuicConfig,
35    task_group: Arc<TaskGroup>,
36) -> Result<H3SendRequest> {
37    let quic_ep = Endpoint::new_quic_addr(SocketAddr::try_from(endpoint.clone())?);
38    let quic_conn = karyon_net::quic::QuicEndpoint::dial(&quic_ep, quic_config.clone()).await?;
39
40    let h3_conn = h3_quinn::Connection::new(quic_conn.inner().clone());
41    let (driver, send_request) = h3::client::new(h3_conn)
42        .await
43        .map_err(|e| Error::HttpError(format!("H3 handshake: {e}")))?;
44
45    // h3 splits I/O from the request API: `driver` polls the QUIC
46    // connection in the background; without it `send_request` would
47    // never make progress. Spawn via the Client task_group so it
48    // gets cancelled on stop.
49    task_group.spawn(driver_task(driver));
50
51    Ok(send_request)
52}
53
54async fn driver_task(mut driver: h3::client::Connection<h3_quinn::Connection, Bytes>) {
55    let closed = futures_util::future::poll_fn(|cx| driver.poll_close(cx)).await;
56    debug!("H3 client driver closed: {closed}");
57}
58
59/// Reads codec frames from the recv half of an HTTP/3 request stream.
60/// One read may hold part of a frame or several frames.
61pub(crate) struct H3Reader<C> {
62    recv: H3RecvStream,
63    buf: ByteBuffer,
64    codec: C,
65}
66
67impl<C: JsonRpcCodec> H3Reader<C> {
68    /// Next message, or `None` when the stream ends.
69    async fn next(&mut self) -> Result<Option<Value>> {
70        loop {
71            if let Some((n, msg)) = self.codec.decode(&mut self.buf)? {
72                self.buf.advance(n);
73                return Ok(Some(msg));
74            }
75            let chunk = self
76                .recv
77                .recv_data()
78                .await
79                .map_err(|e| Error::HttpError(format!("H3 data: {e}")))?;
80            match chunk {
81                Some(chunk) => extend_buf(&mut self.buf, chunk),
82                None => return Ok(None),
83            }
84        }
85    }
86
87    /// Next message as a JSON-RPC response.
88    async fn next_response(&mut self) -> Result<message::Response> {
89        match self.next().await? {
90            Some(msg) => Ok(serde_json::from_value(msg)?),
91            None => Err(Error::HttpError("H3 stream closed before response".into())),
92        }
93    }
94}
95
96/// Send the request headers and `msg`. The request body stays open.
97async fn start_request<C: JsonRpcCodec>(
98    mut send_request: H3SendRequest,
99    config: &HttpConfig,
100    msg: serde_json::Value,
101    codec: &C,
102) -> Result<H3Stream> {
103    let body = encode_msg(codec, &msg)?;
104
105    // The QUIC connection is already established, so the URI's scheme
106    // and authority are only filled in to satisfy `Request::post`'s
107    // parser; the server only routes on `:path`.
108    let req = Request::post("https://localhost/")
109        .header(CONTENT_TYPE, config.content_type.as_str())
110        .body(())
111        .map_err(|e| Error::HttpError(e.to_string()))?;
112
113    let mut stream = send_request
114        .send_request(req)
115        .await
116        .map_err(|e| Error::HttpError(format!("H3 send: {e}")))?;
117
118    stream
119        .send_data(body)
120        .await
121        .map_err(|e| Error::HttpError(format!("H3 data: {e}")))?;
122
123    Ok(stream)
124}
125
126/// Wait for the response headers and split the stream into the send
127/// half and a reader for the reply.
128async fn read_response<C: JsonRpcCodec>(
129    mut stream: H3Stream,
130    codec: &C,
131) -> Result<(H3SendStream, H3Reader<C>)> {
132    let resp = stream
133        .recv_response()
134        .await
135        .map_err(|e| Error::HttpError(format!("H3 recv: {e}")))?;
136
137    if resp.status() != StatusCode::OK {
138        return Err(Error::HttpError(format!("HTTP/3 error: {}", resp.status())));
139    }
140
141    let (send, recv) = stream.split();
142    let reader = H3Reader {
143        recv,
144        buf: ByteBuffer::new(),
145        codec: codec.clone(),
146    };
147    Ok((send, reader))
148}
149
150pub(super) async fn send<C: JsonRpcCodec>(
151    send_request: H3SendRequest,
152    config: &HttpConfig,
153    msg: serde_json::Value,
154    codec: &C,
155) -> Result<message::Response> {
156    let mut stream = start_request(send_request, config, msg, codec).await?;
157    // Finish before reading, so servers that read the whole body first
158    // still answer.
159    stream
160        .finish()
161        .await
162        .map_err(|e| Error::HttpError(format!("H3 finish: {e}")))?;
163    let (_, mut reader) = read_response(stream, codec).await?;
164    reader.next_response().await
165}
166
167/// Subscribe over HTTP/3: send the request with the body left open, read
168/// the initial response, and return the send half and the reader for
169/// notifications.
170pub(super) async fn subscribe<C: JsonRpcCodec>(
171    send_request: H3SendRequest,
172    config: &HttpConfig,
173    msg: serde_json::Value,
174    codec: &C,
175) -> Result<(message::Response, H3SendStream, H3Reader<C>)> {
176    let stream = start_request(send_request, config, msg, codec).await?;
177    let (send, mut reader) = read_response(stream, codec).await?;
178    let rpc_response = reader.next_response().await?;
179    Ok((rpc_response, send, reader))
180}
181
182/// Forward notifications to the subscription until the stream ends or
183/// an unsubscribe request arrives. The unsubscribe goes out on the same
184/// stream, so the server handles it with the subscription's channel.
185// TODO: remove the subscription when this task exits; now `recv()`
186// waits forever once the stream ends.
187pub(super) async fn subscription_task<C: JsonRpcCodec>(
188    mut send: H3SendStream,
189    mut reader: H3Reader<C>,
190    unsub_rx: Receiver<Value>,
191    sub: Arc<Subscription>,
192    decoder: NotificationDecoder,
193) -> Result<()> {
194    loop {
195        match select(reader.next(), unsub_rx.recv()).await {
196            Either::Left(Ok(Some(msg))) => {
197                let nt_res = match decode_notification(decoder, msg) {
198                    Ok(r) => r,
199                    Err(_) => continue,
200                };
201                let val = nt_res.result.unwrap_or(serde_json::json!(""));
202                if sub.notify(val).await.is_err() {
203                    break;
204                }
205            }
206            Either::Left(_) => break,
207            Either::Right(unsub_msg) => {
208                if let Ok(msg) = unsub_msg {
209                    let body = encode_msg(&reader.codec, &msg)?;
210                    let _ = send.send_data(body).await;
211                    let _ = send.finish().await;
212                }
213                break;
214            }
215        }
216    }
217    Ok(())
218}
karyon_jsonrpc/client/http/h3.rs

karyon_jsonrpc/client/http/
h3.rs
