h3_util/
client.rs

1use std::{
2    fmt,
3    future::Future,
4    net::SocketAddr,
5    pin::Pin,
6    task::{Context, Poll},
7};
8
9use futures::future::BoxFuture;
10use hyper::{
11    Request, Response, Uri,
12    body::{Body, Bytes},
13    rt::Executor,
14};
15use tower::{
16    Service,
17    buffer::{Buffer, future::ResponseFuture as BufferResponseFuture},
18    util::BoxService,
19};
20
21use crate::client_conn;
22use crate::{client_body::H3IncomingClient, executor::SharedExec};
23
24const DEFAULT_BUFFER_SIZE: usize = 1024;
25
26pub trait H3Connector: Send + 'static + Clone {
27    type CONN: h3::quic::Connection<
28            Bytes,
29            OpenStreams = Self::OS,
30            SendStream = Self::SS,
31            RecvStream = Self::RS,
32        > + Send;
33    type OS: h3::quic::OpenStreams<Bytes, BidiStream = Self::BS> + Clone + Send; // Clone is needed for cloning send_request
34    type SS: h3::quic::SendStream<Bytes> + Send;
35    type RS: h3::quic::RecvStream + Send;
36    type BS: h3::quic::BidiStream<Bytes, RecvStream = Self::RS, SendStream = Self::SS> + Send;
37
38    fn connect(
39        &self,
40    ) -> impl std::future::Future<Output = Result<Self::CONN, crate::Error>> + std::marker::Send;
41}
42
43/// Use the host:port portion of the uri and resolve to an sockaddr.
44/// If uri host portion is an ip string, then directly use the ip addr without
45/// dns lookup.
46pub async fn dns_resolve(uri: &Uri) -> std::io::Result<Vec<SocketAddr>> {
47    let host_port = uri
48        .authority()
49        .ok_or(std::io::Error::from(std::io::ErrorKind::InvalidInput))?
50        .as_str();
51    match host_port.parse::<SocketAddr>() {
52        Ok(addr) => Ok(vec![addr]),
53        Err(_) => {
54            // uri is using a dns name. try resolve it and return the first.
55            tokio::net::lookup_host(host_port)
56                .await
57                .map(|a| a.collect::<Vec<_>>())
58        }
59    }
60}
61
62/// Cloneable http3 client channel, which can be used to enable multiplexing requests.
63pub struct H3Channel<C, B>
64where
65    C: H3Connector,
66    B: Body + Send + 'static + Unpin,
67    B::Data: Send,
68    B::Error: Into<crate::Error> + Send,
69{
70    #[allow(clippy::type_complexity)]
71    svc: Buffer<
72        Request<B>,
73        BoxFuture<'static, Result<Response<H3IncomingClient<C::RS, Bytes>>, crate::Error>>,
74    >,
75}
76
77impl<C, B> Clone for H3Channel<C, B>
78where
79    C: H3Connector,
80    B: Body + Send + 'static + Unpin,
81    B::Data: Send,
82    B::Error: Into<crate::Error> + Send,
83{
84    fn clone(&self) -> Self {
85        Self {
86            svc: self.svc.clone(),
87        }
88    }
89}
90
91pub struct ResponseFuture<C>
92where
93    C: H3Connector,
94{
95    #[allow(clippy::type_complexity)]
96    inner: BufferResponseFuture<
97        BoxFuture<'static, Result<Response<H3IncomingClient<C::RS, Bytes>>, crate::Error>>,
98    >,
99}
100
101impl<C, B> H3Channel<C, B>
102where
103    C: H3Connector,
104    B: Body + Send + 'static + Unpin,
105    B::Data: Send,
106    B::Error: Into<crate::Error> + Send,
107{
108    pub fn new(connector: C, uri: Uri, executor: Option<SharedExec>) -> Self {
109        let executor = executor.unwrap_or_else(SharedExec::tokio);
110        let svc = H3Connection::new(connector, uri, Some(executor.clone()));
111        let (svc, worker) = Buffer::pair(svc, DEFAULT_BUFFER_SIZE);
112        executor.execute(worker);
113        Self { svc }
114    }
115}
116
117impl<C, B> Service<Request<B>> for H3Channel<C, B>
118where
119    C: H3Connector,
120    B: Body + Send + 'static + Unpin,
121    B::Data: Send,
122    B::Error: Into<crate::Error> + Send,
123{
124    type Response = Response<H3IncomingClient<C::RS, Bytes>>;
125    type Error = crate::Error;
126    type Future = ResponseFuture<C>;
127
128    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
129        Service::poll_ready(&mut self.svc, cx).map_err(crate::Error::from)
130    }
131
132    fn call(&mut self, req: Request<B>) -> Self::Future {
133        let inner = Service::call(&mut self.svc, req);
134        ResponseFuture { inner }
135    }
136}
137
138impl<C> Future for ResponseFuture<C>
139where
140    C: H3Connector,
141{
142    type Output = Result<Response<H3IncomingClient<C::RS, Bytes>>, crate::Error>;
143
144    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
145        Pin::new(&mut self.inner)
146            .poll(cx)
147            .map_err(crate::Error::from)
148    }
149}
150
151impl<C, B> fmt::Debug for H3Channel<C, B>
152where
153    C: H3Connector,
154    B: Body + Send + 'static + Unpin,
155    B::Data: Send,
156    B::Error: Into<crate::Error> + Send,
157{
158    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
159        f.debug_struct("H3Channel").finish()
160    }
161}
162
163impl<C> fmt::Debug for ResponseFuture<C>
164where
165    C: H3Connector,
166{
167    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
168        f.debug_struct("ResponseFuture").finish()
169    }
170}
171
172/// h3 client connection, wrapping inner types for ease of use.
173/// All request will be sent to the connection established using the connector.
174/// Currently connector can only connect to a fixed server (to support grpc use case).
175/// Expand connector to do resolve different server based on uri can be added in future.
176pub struct H3Connection<C, B>
177where
178    C: H3Connector,
179    B: Body + Send + 'static + Unpin,
180    B::Data: Send,
181    B::Error: Into<crate::Error>,
182{
183    #[allow(clippy::type_complexity)]
184    inner: BoxService<Request<B>, Response<H3IncomingClient<C::RS, Bytes>>, crate::Error>,
185}
186
187impl<C, B> H3Connection<C, B>
188where
189    C: H3Connector,
190    B: Body + Send + 'static + Unpin,
191    B::Data: Send,
192    B::Error: Into<crate::Error> + Send,
193{
194    pub fn new(connector: C, uri: Uri, executor: Option<SharedExec>) -> Self {
195        let executor = executor.unwrap_or_else(SharedExec::tokio);
196        let sender = client_conn::RequestSender::new(connector, uri, executor);
197        Self {
198            inner: BoxService::new(sender),
199        }
200    }
201}
202
203impl<C, B> Service<Request<B>> for H3Connection<C, B>
204where
205    C: H3Connector,
206    B: Body + Send + 'static + Unpin,
207    B::Data: Send,
208    B::Error: Into<crate::Error>,
209{
210    type Response = Response<H3IncomingClient<C::RS, Bytes>>;
211    type Error = crate::Error;
212    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
213
214    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
215        Service::poll_ready(&mut self.inner, cx)
216    }
217
218    fn call(&mut self, req: Request<B>) -> Self::Future {
219        self.inner.call(req)
220    }
221}
222
223/// http3 client.
224/// Note the client does not do dns resolve but blindly sends requests
225/// using connections created by the connector.
226/// Used for sending HTTP request directly.
227pub struct H3Client<C, B>
228where
229    C: H3Connector,
230    B: Body + Send + 'static + Unpin,
231    B::Data: Send,
232    B::Error: Into<crate::Error> + Send,
233{
234    channel: H3Connection<C, B>,
235}
236
237impl<C, B> H3Client<C, B>
238where
239    C: H3Connector,
240    B: Body + Send + 'static + Unpin,
241    B::Data: Send,
242    B::Error: Into<crate::Error> + Send,
243{
244    pub fn new(inner: H3Connection<C, B>) -> Self {
245        Self { channel: inner }
246    }
247
248    pub async fn send(
249        &mut self,
250        req: Request<B>,
251    ) -> Result<Response<H3IncomingClient<C::RS, Bytes>>, crate::Error> {
252        // wait for ready
253        futures::future::poll_fn(|cx| self.channel.poll_ready(cx)).await?;
254        self.channel.call(req).await
255    }
256}
h3_util/client.rs

h3_util/
client.rs
