saluki_io/net/client/http/
client.rs

1#[cfg(unix)]
2use std::path::PathBuf;
3use std::{
4    future::Future,
5    pin::Pin,
6    task::{Context, Poll},
7    time::Duration,
8};
9
10use bytes::{Buf, Bytes};
11use http::{Request, Response, Uri};
12use http_body::{Body, Frame, SizeHint};
13use http_body_util::combinators::BoxBody;
14use hyper::body::Incoming;
15use hyper_http_proxy::Proxy;
16use hyper_util::{
17    client::legacy::{connect::capture_connection, Builder},
18    rt::{TokioExecutor, TokioTimer},
19};
20use metrics::Counter;
21use pin_project::pin_project;
22use saluki_error::GenericError;
23use saluki_metrics::MetricsBuilder;
24use saluki_tls::{ClientTLSConfigBuilder, TlsMinimumVersion};
25use stringtheory::MetaString;
26use tower::{timeout::TimeoutLayer, util::BoxCloneService, BoxError, Service, ServiceBuilder, ServiceExt as _};
27
28use super::{
29    conn::{check_connection_state, HttpProtocol, HttpsCapableConnectorBuilder},
30    telemetry::HttpTransactionErrorTelemetry,
31    EndpointTelemetryLayer,
32};
33
34/// The type-erased body type used internally by [`HttpClient`].
35///
36/// All request bodies are converted to this type before being sent over the wire, which ensures a single
37/// monomorphization of the underlying HTTP/2 and TLS stacks regardless of the caller's body type.
38pub type ClientBody = BoxBody<Bytes, Box<dyn std::error::Error + Send + Sync>>;
39
40#[pin_project]
41struct ClientBodyAdapter<B> {
42    #[pin]
43    inner: B,
44    size_hint: SizeHint,
45}
46
47impl<B> Body for ClientBodyAdapter<B>
48where
49    B: Body,
50    B::Data: Buf,
51    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
52{
53    type Data = Bytes;
54    type Error = Box<dyn std::error::Error + Send + Sync>;
55
56    fn poll_frame(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
57        self.project().inner.poll_frame(cx).map(|maybe_frame| {
58            maybe_frame.map(|result| {
59                result
60                    .map(|frame| frame.map_data(|mut data| data.copy_to_bytes(data.remaining())))
61                    .map_err(Into::into)
62            })
63        })
64    }
65
66    fn size_hint(&self) -> SizeHint {
67        self.size_hint.clone()
68    }
69}
70
71/// An HTTP client.
72#[derive(Clone)]
73pub struct HttpClient {
74    inner: BoxCloneService<Request<ClientBody>, Response<Incoming>, BoxError>,
75}
76
77impl HttpClient {
78    /// Creates a new builder for configuring an HTTP client.
79    pub fn builder() -> HttpClientBuilder {
80        HttpClientBuilder::default()
81    }
82
83    /// Sends a request to the server, and waits for a response.
84    ///
85    /// The request body is type-erased internally, so callers can use any body type that implements
86    /// [`Body`] with `Data` types that implement [`Buf`].
87    ///
88    /// # Errors
89    ///
90    /// If there was an error sending the request, an error will be returned.
91    pub async fn send<B>(&mut self, req: Request<B>) -> Result<Response<Incoming>, GenericError>
92    where
93        B: Body + Send + Sync + 'static,
94        B::Data: Buf + Send,
95        B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
96    {
97        let mut req = req.map(into_client_body);
98        let captured_conn = capture_connection(&mut req);
99        let result = self
100            .inner
101            .ready()
102            .await
103            .map_err(GenericError::from_boxed)?
104            .call(req)
105            .await;
106
107        check_connection_state(captured_conn);
108
109        result.map_err(GenericError::from_boxed)
110    }
111}
112
113impl Service<Request<ClientBody>> for HttpClient {
114    type Response = Response<Incoming>;
115    type Error = BoxError;
116    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
117
118    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
119        self.inner.poll_ready(cx)
120    }
121
122    fn call(&mut self, mut req: Request<ClientBody>) -> Self::Future {
123        let captured_conn = capture_connection(&mut req);
124        let fut = self.inner.call(req);
125
126        Box::pin(async move {
127            let result = fut.await;
128
129            check_connection_state(captured_conn);
130
131            result
132        })
133    }
134}
135
136/// Converts an arbitrary body into the type-erased [`ClientBody`].
137///
138/// This uses `Buf::copy_to_bytes` for the data conversion, which is zero-copy when the underlying
139/// data is already `Bytes`.
140pub fn into_client_body<B>(body: B) -> ClientBody
141where
142    B: Body + Send + Sync + 'static,
143    B::Data: Buf + Send,
144    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
145{
146    let size_hint = body.size_hint();
147    BoxBody::new(ClientBodyAdapter { inner: body, size_hint })
148}
149
150/// An HTTP client builder.
151///
152/// Provides an ergonomic builder API for configuring an HTTP client.
153///
154/// # Defaults
155///
156/// A number of sensible defaults are provided:
157///
158/// - support for both HTTP and HTTPS (uses platform's root certificates for server certificate validation)
159/// - support for both HTTP/1.1 and HTTP/2 (automatically negotiated via ALPN)
160/// - non-infinite timeouts for various stages of the request lifecycle (30 second connect timeout, 60 second per-request timeout)
161/// - connection pool for reusing connections (45 second idle connection timeout, and a maximum of 5 idle connections
162///   per host)
163/// - support for FIPS-compliant cryptography if the `fips` feature is enabled in the `saluki-tls` crate
164pub struct HttpClientBuilder {
165    connector_builder: HttpsCapableConnectorBuilder,
166    hyper_builder: Builder,
167    tls_builder: ClientTLSConfigBuilder,
168    request_timeout: Option<Duration>,
169    endpoint_telemetry: Option<EndpointTelemetryLayer>,
170    proxies: Option<Vec<Proxy>>,
171}
172
173impl HttpClientBuilder {
174    /// Sets the timeout when connecting to the remote host.
175    ///
176    /// Defaults to 30 seconds.
177    pub fn with_connect_timeout(mut self, timeout: Duration) -> Self {
178        self.connector_builder = self.connector_builder.with_connect_timeout(timeout);
179        self
180    }
181
182    /// Sets the per-request timeout.
183    ///
184    /// The request timeout applies to each individual request made to the remote host, including each request made when
185    /// retrying a failed request.
186    ///
187    /// Defaults to 20 seconds.
188    pub fn with_request_timeout(mut self, timeout: Duration) -> Self {
189        self.request_timeout = Some(timeout);
190        self
191    }
192
193    /// Allow requests to run indefinitely.
194    ///
195    /// This means there will be no overall timeout for the request, but the request still may be subject to other
196    /// configuration settings, such as the connect timeout or retry policy.
197    pub fn without_request_timeout(mut self) -> Self {
198        self.request_timeout = None;
199        self
200    }
201
202    /// Sets the HTTP protocol selection for client connections.
203    ///
204    /// Defaults to [`HttpProtocol::Auto`], which automatically negotiates HTTP/2 with HTTP/1.1 fallback.
205    pub fn with_http_protocol(mut self, protocol: HttpProtocol) -> Self {
206        self.connector_builder = self.connector_builder.with_http_protocol(protocol);
207        self
208    }
209
210    /// Sets the maximum age of a connection before it's closed.
211    ///
212    /// This is distinct from the maximum idle time: if any connection's age exceeds `limit`, it will be closed rather
213    /// than being reused and added to the idle connection pool.
214    ///
215    /// Defaults to no limit.
216    pub fn with_connection_age_limit<L>(mut self, limit: L) -> Self
217    where
218        L: Into<Option<Duration>>,
219    {
220        self.connector_builder = self.connector_builder.with_connection_age_limit(limit);
221        self
222    }
223
224    /// Sets the maximum number of idle connections per host.
225    ///
226    /// Defaults to 5.
227    pub fn with_max_idle_conns_per_host(mut self, max: usize) -> Self {
228        self.hyper_builder.pool_max_idle_per_host(max);
229        self
230    }
231
232    /// Sets the idle connection timeout.
233    ///
234    /// Once a connection has been idle in the pool for longer than this duration, it will be closed and removed from
235    /// the pool.
236    ///
237    /// Defaults to 45 seconds.
238    pub fn with_idle_conn_timeout(mut self, timeout: Duration) -> Self {
239        self.hyper_builder.pool_idle_timeout(timeout);
240        self
241    }
242
243    /// Sets the proxies to be used for outgoing requests.
244    ///
245    /// Defaults to no proxies. (i.e requests will be sent directly without using a proxy).
246    pub fn with_proxies(mut self, proxies: Vec<Proxy>) -> Self {
247        self.proxies = Some(proxies);
248        self
249    }
250
251    /// Enables per-endpoint telemetry for HTTP transactions.
252    ///
253    /// See [`EndpointTelemetryLayer`] for more information.
254    pub fn with_endpoint_telemetry<F>(mut self, metrics_builder: MetricsBuilder, endpoint_name_fn: Option<F>) -> Self
255    where
256        F: Fn(&Uri) -> Option<MetaString> + Send + Sync + 'static,
257    {
258        let error_telemetry = HttpTransactionErrorTelemetry::from_builder(&metrics_builder);
259        self.connector_builder = self.connector_builder.with_error_telemetry(error_telemetry.clone());
260
261        let mut layer = EndpointTelemetryLayer::default()
262            .with_metrics_builder(metrics_builder)
263            .with_error_telemetry(error_telemetry);
264
265        if let Some(endpoint_name_fn) = endpoint_name_fn {
266            layer = layer.with_endpoint_name_fn(endpoint_name_fn);
267        }
268
269        self.endpoint_telemetry = Some(layer);
270        self
271    }
272
273    /// Sets a Unix domain socket path to route all connections through.
274    ///
275    /// When set, the client will connect to this Unix socket instead of performing DNS resolution
276    /// and TCP connection. The URI host is ignored—all requests are sent through the configured
277    /// socket.
278    ///
279    /// Defaults to unset (TCP connections via DNS).
280    #[cfg(unix)]
281    pub fn with_unix_socket_path<P: Into<PathBuf>>(mut self, path: P) -> Self {
282        self.connector_builder = self.connector_builder.with_unix_socket_path(path);
283        self
284    }
285
286    /// Sets the TLS configuration.
287    ///
288    /// A TLS configuration builder is provided to allow for more advanced configuration of the TLS connection.
289    pub fn with_tls_config<F>(mut self, f: F) -> Self
290    where
291        F: FnOnce(ClientTLSConfigBuilder) -> ClientTLSConfigBuilder,
292    {
293        self.tls_builder = f(self.tls_builder);
294        self
295    }
296
297    /// Sets the minimum TLS protocol version for HTTPS connections.
298    ///
299    /// Defaults to TLS 1.2.
300    ///
301    /// This updates the same TLS builder configured by [`Self::with_tls_config`], so call order matters when both
302    /// methods change the minimum TLS version.
303    pub fn with_min_tls_version(mut self, version: TlsMinimumVersion) -> Self {
304        self.tls_builder = self.tls_builder.with_min_tls_version(version);
305        self
306    }
307
308    /// Sets the underlying Hyper client configuration.
309    ///
310    /// This is provided to allow for more advanced configuration of the Hyper client itself, and should generally be
311    /// used sparingly.
312    pub fn with_hyper_config<F>(mut self, f: F) -> Self
313    where
314        F: FnOnce(&mut Builder),
315    {
316        f(&mut self.hyper_builder);
317        self
318    }
319
320    /// Sets a counter that gets incremented with the number of bytes sent over the connection.
321    ///
322    /// This tracks bytes sent at the HTTP client level, which includes headers and body but doesn't include underlying
323    /// transport overhead, such as TLS handshaking, and so on.
324    ///
325    /// Defaults to unset.
326    pub fn with_bytes_sent_counter(mut self, counter: Counter) -> Self {
327        self.connector_builder = self.connector_builder.with_bytes_sent_counter(counter);
328        self
329    }
330
331    /// Builds the `HttpClient`.
332    ///
333    /// # Errors
334    ///
335    /// If there was an error building the TLS configuration for the client, an error will be returned.
336    pub fn build(self) -> Result<HttpClient, GenericError> {
337        let tls_config = self.tls_builder.build()?;
338        let connector = self.connector_builder.build(tls_config)?;
339        // TODO(fips): Look into updating `hyper-http-proxy` to use the provided connector for establishing the
340        // connection to the proxy itself, even when the proxy is at an HTTPS URL, to ensure our desired TLS stack is
341        // being used.
342        let mut proxy_connector = hyper_http_proxy::ProxyConnector::new(connector)?;
343        if let Some(proxies) = &self.proxies {
344            for proxy in proxies {
345                proxy_connector.add_proxy(proxy.to_owned());
346            }
347        }
348        let client = self.hyper_builder.build(proxy_connector);
349
350        let inner = ServiceBuilder::new()
351            .option_layer(self.request_timeout.map(TimeoutLayer::new))
352            .option_layer(self.endpoint_telemetry)
353            .service(client.map_err(BoxError::from))
354            .boxed_clone();
355
356        Ok(HttpClient { inner })
357    }
358}
359
360impl Default for HttpClientBuilder {
361    fn default() -> Self {
362        let mut hyper_builder = Builder::new(TokioExecutor::new());
363        hyper_builder
364            .pool_timer(TokioTimer::new())
365            .pool_max_idle_per_host(5)
366            .pool_idle_timeout(Duration::from_secs(45));
367
368        Self {
369            connector_builder: HttpsCapableConnectorBuilder::default(),
370            hyper_builder,
371            tls_builder: ClientTLSConfigBuilder::new(),
372            request_timeout: Some(Duration::from_secs(20)),
373            endpoint_telemetry: None,
374            proxies: None,
375        }
376    }
377}
378
379#[cfg(test)]
380mod tests {
381    use http_body::Body as _;
382    use http_body_util::Full;
383
384    use super::*;
385
386    #[test]
387    fn into_client_body_preserves_exact_size_hint() {
388        let body = Full::new(Bytes::from_static(b"hello"));
389        let converted = into_client_body(body);
390
391        assert_eq!(Some(5), converted.size_hint().exact());
392    }
393}
saluki_io/net/client/http/client.rs

saluki_io/net/client/http/
client.rs
