Skip to main content

aws_smithy_http_client/
client.rs

1/*
2 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3 * SPDX-License-Identifier: Apache-2.0
4 */
5
6mod dns;
7/// Proxy configuration
8pub mod proxy;
9mod timeout;
10/// TLS connector(s)
11pub mod tls;
12
13pub(crate) mod connect;
14
15use crate::cfg::cfg_tls;
16use crate::tls::TlsContext;
17use aws_smithy_async::future::timeout::TimedOutError;
18use aws_smithy_async::rt::sleep::{default_async_sleep, AsyncSleep, SharedAsyncSleep};
19use aws_smithy_runtime_api::box_error::BoxError;
20use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
21use aws_smithy_runtime_api::client::connection::ConnectionMetadata;
22use aws_smithy_runtime_api::client::connector_metadata::ConnectorMetadata;
23use aws_smithy_runtime_api::client::http::{
24    HttpClient, HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpClient,
25    SharedHttpConnector,
26};
27use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};
28use aws_smithy_runtime_api::client::result::ConnectorError;
29use aws_smithy_runtime_api::client::runtime_components::{
30    RuntimeComponents, RuntimeComponentsBuilder,
31};
32use aws_smithy_runtime_api::shared::IntoShared;
33use aws_smithy_types::body::SdkBody;
34use aws_smithy_types::config_bag::ConfigBag;
35use aws_smithy_types::error::display::DisplayErrorContext;
36use aws_smithy_types::retry::ErrorKind;
37use client::connect::Connection;
38use h2::Reason;
39use http_1x::{Extensions, Uri};
40use hyper::rt::{Read, Write};
41use hyper_util::client::legacy as client;
42use hyper_util::client::legacy::connect::dns::GaiResolver;
43use hyper_util::client::legacy::connect::{
44    capture_connection, CaptureConnection, Connect, HttpConnector as HyperHttpConnector, HttpInfo,
45};
46use hyper_util::client::proxy::matcher::Matcher;
47use hyper_util::rt::{TokioExecutor, TokioTimer};
48use std::borrow::Cow;
49use std::collections::HashMap;
50use std::error::Error;
51use std::fmt;
52use std::sync::RwLock;
53use std::time::Duration;
54
55/// Given `HttpConnectorSettings` and an `SharedAsyncSleep`, create a `SharedHttpConnector` from defaults depending on what cargo features are activated.
56pub fn default_connector(
57    settings: &HttpConnectorSettings,
58    sleep: Option<SharedAsyncSleep>,
59) -> Option<SharedHttpConnector> {
60    #[cfg(feature = "rustls-aws-lc")]
61    {
62        tracing::trace!(settings = ?settings, sleep = ?sleep, "creating a new default connector");
63        let mut conn_builder = Connector::builder().connector_settings(settings.clone());
64
65        if let Some(sleep) = sleep {
66            conn_builder = conn_builder.sleep_impl(sleep);
67        }
68
69        let conn = conn_builder
70            .tls_provider(tls::Provider::Rustls(
71                tls::rustls_provider::CryptoMode::AwsLc,
72            ))
73            .build();
74        Some(SharedHttpConnector::new(conn))
75    }
76    #[cfg(not(feature = "rustls-aws-lc"))]
77    {
78        tracing::trace!(settings = ?settings, sleep = ?sleep, "no default connector available");
79        None
80    }
81}
82
83/// [`HttpConnector`] used to make HTTP requests.
84///
85/// This connector also implements socket connect and read timeouts.
86///
87/// This shouldn't be used directly in most cases.
88/// See the docs on [`Builder`] for examples of how to customize the HTTP client.
89#[derive(Debug)]
90pub struct Connector {
91    adapter: Box<dyn HttpConnector>,
92}
93
94impl Connector {
95    /// Builder for an HTTP connector.
96    pub fn builder() -> ConnectorBuilder {
97        ConnectorBuilder::default()
98    }
99}
100
101impl HttpConnector for Connector {
102    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
103        self.adapter.call(request)
104    }
105}
106
107/// Builder for [`Connector`].
108#[derive(Debug, Clone)]
109pub struct ConnectorBuilder<Tls = TlsUnset> {
110    connector_settings: Option<HttpConnectorSettings>,
111    sleep_impl: Option<SharedAsyncSleep>,
112    client_builder: Option<hyper_util::client::legacy::Builder>,
113    pool_idle_timeout: Option<Option<Duration>>,
114    pool_max_idle_per_host: Option<usize>,
115    enable_tcp_nodelay: bool,
116    interface: Option<String>,
117    proxy_config: Option<proxy::ProxyConfig>,
118    #[allow(unused)]
119    tls: Tls,
120}
121
122impl<Tls: Default> Default for ConnectorBuilder<Tls> {
123    fn default() -> Self {
124        Self {
125            connector_settings: None,
126            sleep_impl: None,
127            client_builder: None,
128            pool_idle_timeout: None,
129            pool_max_idle_per_host: None,
130            // Curated default: TCP_NODELAY on. Without it, Nagle's algorithm
131            // can hold a small write while earlier data is unacknowledged. On
132            // request shapes emitted as multiple sub-MSS writes, this can add
133            // an ACK wait, often RTT plus delayed-ACK time. Opt out with
134            // `enable_tcp_nodelay(false)`.
135            enable_tcp_nodelay: true,
136            interface: None,
137            proxy_config: None,
138            tls: Tls::default(),
139        }
140    }
141}
142
143/// Initial builder state, `TlsProvider` choice required
144#[derive(Default, Debug, Clone)]
145#[non_exhaustive]
146pub struct TlsUnset {}
147
148/// TLS implementation selected
149#[derive(Debug, Clone)]
150pub struct TlsProviderSelected {
151    #[allow(unused)]
152    provider: tls::Provider,
153    #[allow(unused)]
154    context: TlsContext,
155}
156
157impl ConnectorBuilder<TlsUnset> {
158    /// Set the TLS implementation to use for this connector
159    pub fn tls_provider(self, provider: tls::Provider) -> ConnectorBuilder<TlsProviderSelected> {
160        ConnectorBuilder {
161            connector_settings: self.connector_settings,
162            sleep_impl: self.sleep_impl,
163            client_builder: self.client_builder,
164            enable_tcp_nodelay: self.enable_tcp_nodelay,
165            interface: self.interface,
166            proxy_config: self.proxy_config,
167            pool_idle_timeout: self.pool_idle_timeout,
168            pool_max_idle_per_host: self.pool_max_idle_per_host,
169            tls: TlsProviderSelected {
170                provider,
171                context: TlsContext::default(),
172            },
173        }
174    }
175
176    /// Build an HTTP connector sans TLS
177    #[doc(hidden)]
178    pub fn build_http(self) -> Connector {
179        if let Some(ref proxy_config) = self.proxy_config {
180            if proxy_config.requires_tls() {
181                tracing::warn!(
182                    "HTTPS proxy configured but no TLS provider set. \
183                     Connections to HTTPS proxy servers will fail. \
184                     Consider configuring a TLS provider to enable TLS support."
185                );
186            }
187        }
188
189        let base = self.base_connector();
190
191        // Wrap with HTTP proxy support if proxy is configured
192        let proxy_config = self
193            .proxy_config
194            .clone()
195            .unwrap_or_else(proxy::ProxyConfig::disabled);
196
197        if !proxy_config.is_disabled() {
198            let http_proxy_connector = connect::HttpProxyConnector::new(base, proxy_config);
199            self.wrap_connector(http_proxy_connector)
200        } else {
201            self.wrap_connector(base)
202        }
203    }
204}
205
206impl<Any> ConnectorBuilder<Any> {
207    /// Create a [`Connector`] from this builder and a given connector.
208    pub(crate) fn wrap_connector<C>(self, tcp_connector: C) -> Connector
209    where
210        C: Send + Sync + 'static,
211        C: Clone,
212        C: tower::Service<Uri>,
213        C::Response: Read + Write + Connection + Send + Sync + Unpin,
214        C: Connect,
215        C::Future: Unpin + Send + 'static,
216        C::Error: Into<BoxError>,
217    {
218        let client_builder = self.client_builder.unwrap_or_else(|| {
219            new_tokio_hyper_builder(self.pool_idle_timeout, self.pool_max_idle_per_host)
220        });
221        let sleep_impl = self.sleep_impl.or_else(default_async_sleep);
222        let (connect_timeout, read_timeout) = self
223            .connector_settings
224            .map(|c| (c.connect_timeout(), c.read_timeout()))
225            .unwrap_or((None, None));
226
227        let connector = match connect_timeout {
228            Some(duration) => timeout::ConnectTimeout::new(
229                tcp_connector,
230                sleep_impl
231                    .clone()
232                    .expect("a sleep impl must be provided in order to have a connect timeout"),
233                duration,
234            ),
235            None => timeout::ConnectTimeout::no_timeout(tcp_connector),
236        };
237        let base = client_builder.build(connector);
238        let read_timeout = match read_timeout {
239            Some(duration) => timeout::HttpReadTimeout::new(
240                base,
241                sleep_impl.expect("a sleep impl must be provided in order to have a read timeout"),
242                duration,
243            ),
244            None => timeout::HttpReadTimeout::no_timeout(base),
245        };
246
247        let proxy_matcher = self
248            .proxy_config
249            .as_ref()
250            .map(|config| config.clone().into_hyper_util_matcher());
251
252        Connector {
253            adapter: Box::new(Adapter {
254                client: read_timeout,
255                proxy_matcher,
256            }),
257        }
258    }
259
260    /// Get the base TCP connector by mapping our config to the underlying `HttpConnector` from hyper
261    /// (which is a base TCP connector with no TLS or any wrapping)
262    fn base_connector(&self) -> HyperHttpConnector {
263        self.base_connector_with_resolver(GaiResolver::new())
264    }
265
266    /// Get the base TCP connector by mapping our config to the underlying `HttpConnector` from hyper
267    /// using the given resolver `R`
268    fn base_connector_with_resolver<R>(&self, resolver: R) -> HyperHttpConnector<R> {
269        let mut conn = HyperHttpConnector::new_with_resolver(resolver);
270        conn.set_nodelay(self.enable_tcp_nodelay);
271        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
272        if let Some(interface) = &self.interface {
273            conn.set_interface(interface);
274        }
275        conn
276    }
277
278    /// Set the async sleep implementation used for timeouts
279    ///
280    /// Calling this is only necessary for testing or to use something other than
281    /// [`default_async_sleep`].
282    pub fn sleep_impl(mut self, sleep_impl: impl AsyncSleep + 'static) -> Self {
283        self.sleep_impl = Some(sleep_impl.into_shared());
284        self
285    }
286
287    /// Set the async sleep implementation used for timeouts
288    ///
289    /// Calling this is only necessary for testing or to use something other than
290    /// [`default_async_sleep`].
291    pub fn set_sleep_impl(&mut self, sleep_impl: Option<SharedAsyncSleep>) -> &mut Self {
292        self.sleep_impl = sleep_impl;
293        self
294    }
295
296    /// Configure the HTTP settings for the `HyperAdapter`
297    pub fn connector_settings(mut self, connector_settings: HttpConnectorSettings) -> Self {
298        self.connector_settings = Some(connector_settings);
299        self
300    }
301
302    /// Configure the HTTP settings for the `HyperAdapter`
303    pub fn set_connector_settings(
304        &mut self,
305        connector_settings: Option<HttpConnectorSettings>,
306    ) -> &mut Self {
307        self.connector_settings = connector_settings;
308        self
309    }
310
311    /// Configure `SO_NODELAY` for all sockets to the supplied value `nodelay`
312    pub fn enable_tcp_nodelay(mut self, nodelay: bool) -> Self {
313        self.enable_tcp_nodelay = nodelay;
314        self
315    }
316
317    /// Configure `SO_NODELAY` for all sockets to the supplied value `nodelay`
318    pub fn set_enable_tcp_nodelay(&mut self, nodelay: bool) -> &mut Self {
319        self.enable_tcp_nodelay = nodelay;
320        self
321    }
322
323    /// Sets the value for the `SO_BINDTODEVICE` option on this socket.
324    ///
325    /// If a socket is bound to an interface, only packets received from that particular
326    /// interface are processed by the socket. Note that this only works for some socket
327    /// types (e.g. `AF_INET` sockets).
328    ///
329    /// On Linux it can be used to specify a [VRF], but the binary needs to either have
330    /// `CAP_NET_RAW` capability set or be run as root.
331    ///
332    /// This function is only available on Android, Fuchsia, and Linux.
333    ///
334    /// [VRF]: https://www.kernel.org/doc/Documentation/networking/vrf.txt
335    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
336    pub fn set_interface<S: Into<String>>(&mut self, interface: S) -> &mut Self {
337        self.interface = Some(interface.into());
338        self
339    }
340
341    /// Configure proxy settings for this connector
342    ///
343    /// This method allows you to set explicit proxy configuration for the HTTP client.
344    /// The proxy configuration will be used to determine whether requests should be
345    /// routed through a proxy server or connect directly.
346    ///
347    /// # Examples
348    ///
349    /// ```rust
350    /// # #[cfg(feature = "rustls-aws-lc")]
351    /// # {
352    /// use aws_smithy_http_client::{Connector, proxy::ProxyConfig, tls};
353    ///
354    /// let proxy_config = ProxyConfig::http("http://proxy.example.com:8080")?;
355    /// let connector = Connector::builder()
356    ///     .proxy_config(proxy_config)
357    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
358    ///     .build();
359    /// # }
360    /// # Ok::<(), Box<dyn std::error::Error>>(())
361    /// ```
362    pub fn proxy_config(mut self, config: proxy::ProxyConfig) -> Self {
363        self.proxy_config = Some(config);
364        self
365    }
366
367    /// Configure proxy settings for this connector
368    ///
369    /// This is the mutable version of [`proxy_config`](Self::proxy_config).
370    pub fn set_proxy_config(&mut self, config: Option<proxy::ProxyConfig>) -> &mut Self {
371        self.proxy_config = config;
372        self
373    }
374
375    /// Set an optional timeout for idle sockets being kept-alive.
376    ///
377    /// Pass `None` to disable timeout.
378    ///
379    /// Defaults to Hyper's default timeout, which is currently 90 seconds - see
380    /// [hyper_util::client::legacy::Builder::pool_idle_timeout],
381    /// but unlike that function, there is no need to call `pool_timer` yourself.
382    ///
383    /// # Examples
384    ///
385    /// ```rust
386    /// # #[cfg(feature = "rustls-aws-lc")]
387    /// # {
388    /// use aws_smithy_http_client::{Connector, tls};
389    /// use std::time::Duration;
390    ///
391    /// let connector = Connector::builder()
392    ///     .pool_idle_timeout(Duration::from_secs(30))
393    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
394    ///     .build();
395    /// # }
396    /// # Ok::<(), Box<dyn std::error::Error>>(())
397    /// ```
398    pub fn pool_idle_timeout<D>(mut self, val: D) -> Self
399    where
400        D: Into<Option<Duration>>,
401    {
402        self.pool_idle_timeout = Some(val.into());
403        self
404    }
405
406    /// Set an optional timeout for idle sockets being kept-alive.
407    ///
408    /// Pass `None` to use Hyper's default timeout, `Some(None)` to disable timeouts.
409    ///
410    /// This is the mutable version of [`pool_idle_timeout`](Self::pool_idle_timeout).
411    ///
412    /// # Examples
413    ///
414    /// ```rust
415    /// # #[cfg(feature = "rustls-aws-lc")]
416    /// # {
417    /// use aws_smithy_http_client::{Connector, tls};
418    /// use std::time::Duration;
419    ///
420    /// let mut connector = Connector::builder();
421    /// connector
422    ///     .set_pool_idle_timeout(Some(Some(Duration::from_secs(30))));
423    /// connector
424    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
425    ///     .build();
426    /// # }
427    /// # Ok::<(), Box<dyn std::error::Error>>(())
428    /// ```
429    pub fn set_pool_idle_timeout(&mut self, val: Option<Option<Duration>>) -> &mut Self {
430        self.pool_idle_timeout = val;
431        self
432    }
433
434    /// Sets the maximum number of idle pooled connections allowed per host.
435    ///
436    /// Default is determined by the underlying hyper client, which is currently no limit
437    /// (`usize::MAX`) - see [hyper_util::client::legacy::Builder::pool_max_idle_per_host].
438    ///
439    /// # Examples
440    ///
441    /// ```rust
442    /// # #[cfg(feature = "rustls-aws-lc")]
443    /// # {
444    /// use aws_smithy_http_client::{Connector, tls};
445    ///
446    /// let connector = Connector::builder()
447    ///     .pool_max_idle_per_host(70)
448    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
449    ///     .build();
450    /// # }
451    /// # Ok::<(), Box<dyn std::error::Error>>(())
452    /// ```
453    pub fn pool_max_idle_per_host(mut self, val: usize) -> Self {
454        self.pool_max_idle_per_host = Some(val);
455        self
456    }
457
458    /// Sets the maximum number of idle pooled connections allowed per host.
459    ///
460    /// Pass `None` to use the hyper default (currently no limit, `usize::MAX`) - see
461    /// [hyper_util::client::legacy::Builder::pool_max_idle_per_host].
462    ///
463    /// This is the mutable version of [`pool_max_idle_per_host`](Self::pool_max_idle_per_host).
464    pub fn set_pool_max_idle_per_host(&mut self, val: Option<usize>) -> &mut Self {
465        self.pool_max_idle_per_host = val;
466        self
467    }
468
469    /// Override the Hyper client [`Builder`](hyper_util::client::legacy::Builder) used to construct this client.
470    ///
471    /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.
472    pub(crate) fn hyper_builder(
473        mut self,
474        hyper_builder: hyper_util::client::legacy::Builder,
475    ) -> Self {
476        self.set_hyper_builder(Some(hyper_builder));
477        self
478    }
479
480    /// Override the Hyper client [`Builder`](hyper_util::client::legacy::Builder) used to construct this client.
481    ///
482    /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.
483    pub(crate) fn set_hyper_builder(
484        &mut self,
485        hyper_builder: Option<hyper_util::client::legacy::Builder>,
486    ) -> &mut Self {
487        self.client_builder = hyper_builder;
488        self
489    }
490}
491
492/// Adapter to use a Hyper 1.0-based Client as an `HttpConnector`
493///
494/// This adapter also enables TCP `CONNECT` and HTTP `READ` timeouts via [`Connector::builder`].
495struct Adapter<C> {
496    client: timeout::HttpReadTimeout<
497        hyper_util::client::legacy::Client<timeout::ConnectTimeout<C>, SdkBody>,
498    >,
499    proxy_matcher: Option<Matcher>,
500}
501
502impl<C> fmt::Debug for Adapter<C> {
503    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
504        f.debug_struct("Adapter")
505            .field("client", &"** hyper client **")
506            .field("proxy_matcher", &self.proxy_matcher.is_some())
507            .finish()
508    }
509}
510
511/// Extract a smithy connection from a hyper CaptureConnection
512fn extract_smithy_connection(capture_conn: &CaptureConnection) -> Option<ConnectionMetadata> {
513    let capture_conn = capture_conn.clone();
514    if let Some(conn) = capture_conn.clone().connection_metadata().as_ref() {
515        let mut extensions = Extensions::new();
516        conn.get_extras(&mut extensions);
517        let http_info = extensions.get::<HttpInfo>();
518        let mut builder = ConnectionMetadata::builder()
519            .proxied(conn.is_proxied())
520            .poison_fn(move || match capture_conn.connection_metadata().as_ref() {
521                Some(conn) => conn.poison(),
522                None => tracing::trace!("no connection existed to poison"),
523            });
524
525        builder
526            .set_local_addr(http_info.map(|info| info.local_addr()))
527            .set_remote_addr(http_info.map(|info| info.remote_addr()));
528
529        let smithy_connection = builder.build();
530
531        Some(smithy_connection)
532    } else {
533        None
534    }
535}
536
537fn new_tokio_hyper_builder(
538    pool_idle_timeout: Option<Option<Duration>>,
539    pool_max_idle_per_host: Option<usize>,
540) -> hyper_util::client::legacy::Builder {
541    let mut builder = hyper_util::client::legacy::Builder::new(TokioExecutor::new());
542    // Explicitly setting the pool_timer is required for connection timeouts to work.
543    builder.pool_timer(TokioTimer::new());
544
545    if let Some(pool_idle_timeout) = pool_idle_timeout {
546        builder.pool_idle_timeout(pool_idle_timeout);
547    }
548
549    if let Some(max_idle) = pool_max_idle_per_host {
550        builder.pool_max_idle_per_host(max_idle);
551    }
552
553    builder
554}
555
556impl<C> Adapter<C> {
557    /// Add proxy authentication header to the request if needed
558    fn add_proxy_auth_header(&self, request: &mut http_1x::Request<SdkBody>) {
559        // Only add auth for HTTP requests (not HTTPS which uses CONNECT tunneling)
560        if request.uri().scheme() != Some(&http_1x::uri::Scheme::HTTP) {
561            return;
562        }
563
564        // Don't override existing proxy authorization header
565        if request
566            .headers()
567            .contains_key(http_1x::header::PROXY_AUTHORIZATION)
568        {
569            return;
570        }
571
572        if let Some(ref matcher) = self.proxy_matcher {
573            if let Some(intercept) = matcher.intercept(request.uri()) {
574                // Add basic auth header if available
575                if let Some(auth_header) = intercept.basic_auth() {
576                    request
577                        .headers_mut()
578                        .insert(http_1x::header::PROXY_AUTHORIZATION, auth_header.clone());
579                    tracing::debug!("added proxy authentication header for {}", request.uri());
580                }
581            }
582        }
583    }
584}
585
586impl<C> HttpConnector for Adapter<C>
587where
588    C: Clone + Send + Sync + 'static,
589    C: tower::Service<Uri>,
590    C::Response: Connection + Read + Write + Unpin + 'static,
591    timeout::ConnectTimeout<C>: Connect,
592    C::Future: Unpin + Send + 'static,
593    C::Error: Into<BoxError>,
594{
595    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
596        let mut request = match request.try_into_http1x() {
597            Ok(request) => request,
598            Err(err) => {
599                return HttpConnectorFuture::ready(Err(ConnectorError::user(err.into())));
600            }
601        };
602
603        self.add_proxy_auth_header(&mut request);
604
605        let capture_connection = capture_connection(&mut request);
606        if let Some(capture_smithy_connection) =
607            request.extensions().get::<CaptureSmithyConnection>()
608        {
609            capture_smithy_connection
610                .set_connection_retriever(move || extract_smithy_connection(&capture_connection));
611        }
612        let mut client = self.client.clone();
613        use tower::Service;
614        let fut = client.call(request);
615        HttpConnectorFuture::new(async move {
616            let response = fut
617                .await
618                .map_err(downcast_error)?
619                .map(SdkBody::from_body_1_x);
620            match HttpResponse::try_from(response) {
621                Ok(response) => Ok(response),
622                Err(err) => Err(ConnectorError::other(err.into(), None)),
623            }
624        })
625    }
626}
627
628/// Downcast errors coming out of hyper into an appropriate `ConnectorError`
629fn downcast_error(err: BoxError) -> ConnectorError {
630    // is a `TimedOutError` (from aws_smithy_async::timeout) in the chain? if it is, this is a timeout
631    if find_source::<TimedOutError>(err.as_ref()).is_some() {
632        return ConnectorError::timeout(err);
633    }
634    // is the top of chain error actually already a `ConnectorError`? return that directly
635    let err = match err.downcast::<ConnectorError>() {
636        Ok(connector_error) => return *connector_error,
637        Err(box_error) => box_error,
638    };
639    // generally, the top of chain will probably be a hyper error. Go through a set of hyper specific
640    // error classifications
641    let err = match find_source::<hyper::Error>(err.as_ref()) {
642        Some(hyper_error) => return to_connector_error(hyper_error)(err),
643        None => match find_source::<hyper_util::client::legacy::Error>(err.as_ref()) {
644            Some(hyper_util_err) => {
645                if hyper_util_err.is_connect()
646                    || find_source::<std::io::Error>(hyper_util_err).is_some()
647                {
648                    return ConnectorError::io(err);
649                }
650                err
651            }
652            None => err,
653        },
654    };
655
656    // otherwise, we have no idea!
657    ConnectorError::other(err, None)
658}
659
660/// Convert a [`hyper::Error`] into a [`ConnectorError`]
661fn to_connector_error(err: &hyper::Error) -> fn(BoxError) -> ConnectorError {
662    if err.is_timeout() || find_source::<timeout::HttpTimeoutError>(err).is_some() {
663        return ConnectorError::timeout;
664    }
665    if err.is_user() {
666        return ConnectorError::user;
667    }
668    if err.is_closed() || err.is_canceled() || find_source::<std::io::Error>(err).is_some() {
669        return ConnectorError::io;
670    }
671    // We sometimes receive this from S3: hyper::Error(IncompleteMessage)
672    if err.is_incomplete_message() {
673        return |err: BoxError| ConnectorError::other(err, Some(ErrorKind::TransientError));
674    }
675
676    if let Some(h2_err) = find_source::<h2::Error>(err) {
677        if h2_err.is_go_away()
678            || (h2_err.is_reset() && h2_err.reason() == Some(Reason::REFUSED_STREAM))
679        {
680            return ConnectorError::io;
681        }
682    }
683
684    tracing::warn!(err = %DisplayErrorContext(&err), "unrecognized error from Hyper. If this error should be retried, please file an issue.");
685    |err: BoxError| ConnectorError::other(err, None)
686}
687
688fn find_source<'a, E: Error + 'static>(err: &'a (dyn Error + 'static)) -> Option<&'a E> {
689    let mut next = Some(err);
690    while let Some(err) = next {
691        if let Some(matching_err) = err.downcast_ref::<E>() {
692            return Some(matching_err);
693        }
694        next = err.source();
695    }
696    None
697}
698
699// TODO(https://github.com/awslabs/aws-sdk-rust/issues/1090): CacheKey must also include ptr equality to any
700// runtime components that are used—sleep_impl as a base (unless we prohibit overriding sleep impl)
701// If we decide to put a DnsResolver in RuntimeComponents, then we'll need to handle that as well.
702#[derive(Clone, Debug, Eq, PartialEq, Hash)]
703struct CacheKey {
704    connect_timeout: Option<Duration>,
705    read_timeout: Option<Duration>,
706}
707
708impl From<&HttpConnectorSettings> for CacheKey {
709    fn from(value: &HttpConnectorSettings) -> Self {
710        Self {
711            connect_timeout: value.connect_timeout(),
712            read_timeout: value.read_timeout(),
713        }
714    }
715}
716
717struct HyperClient<F> {
718    connector_cache: RwLock<HashMap<CacheKey, SharedHttpConnector>>,
719    client_builder: hyper_util::client::legacy::Builder,
720    connector_fn: F,
721}
722
723impl<F> fmt::Debug for HyperClient<F> {
724    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
725        f.debug_struct("HyperClient")
726            .field("connector_cache", &self.connector_cache)
727            .field("client_builder", &self.client_builder)
728            .finish()
729    }
730}
731
732impl<F> HttpClient for HyperClient<F>
733where
734    F: Fn(
735            hyper_util::client::legacy::Builder,
736            Option<&HttpConnectorSettings>,
737            Option<&RuntimeComponents>,
738        ) -> Connector
739        + Send
740        + Sync
741        + 'static,
742{
743    fn http_connector(
744        &self,
745        settings: &HttpConnectorSettings,
746        components: &RuntimeComponents,
747    ) -> SharedHttpConnector {
748        let key = CacheKey::from(settings);
749        let mut connector = self.connector_cache.read().unwrap().get(&key).cloned();
750        if connector.is_none() {
751            let mut cache = self.connector_cache.write().unwrap();
752            // Short-circuit if another thread already wrote a connector to the cache for this key
753            if !cache.contains_key(&key) {
754                let start = components.time_source().map(|ts| ts.now());
755                let connector = (self.connector_fn)(
756                    self.client_builder.clone(),
757                    Some(settings),
758                    Some(components),
759                );
760                let end = components.time_source().map(|ts| ts.now());
761                if let (Some(start), Some(end)) = (start, end) {
762                    if let Ok(elapsed) = end.duration_since(start) {
763                        tracing::debug!("new connector created in {:?}", elapsed);
764                    }
765                }
766                let connector = SharedHttpConnector::new(connector);
767                cache.insert(key.clone(), connector);
768            }
769            connector = cache.get(&key).cloned();
770        }
771
772        connector.expect("cache populated above")
773    }
774
775    fn validate_base_client_config(
776        &self,
777        _: &RuntimeComponentsBuilder,
778        _: &ConfigBag,
779    ) -> Result<(), BoxError> {
780        // Initialize the TCP connector at this point so that native certs load
781        // at client initialization time instead of upon first request. We do it
782        // here rather than at construction so that it won't run if this is not
783        // the selected HTTP client for the base config (for example, if this was
784        // the default HTTP client, and it was overridden by a later plugin).
785        let _ = (self.connector_fn)(self.client_builder.clone(), None, None);
786        Ok(())
787    }
788
789    fn connector_metadata(&self) -> Option<ConnectorMetadata> {
790        Some(ConnectorMetadata::new("hyper", Some(Cow::Borrowed("1.x"))))
791    }
792}
793
794/// Builder for a hyper-backed [`HttpClient`] implementation.
795///
796/// This builder can be used to customize the underlying TCP connector used, as well as
797/// hyper client configuration.
798///
799/// # Examples
800///
801/// Construct a Hyper client with the RusTLS TLS implementation.
802/// This can be useful when you want to share a Hyper connector between multiple
803/// generated Smithy clients.
804#[derive(Clone, Default, Debug)]
805pub struct Builder<Tls = TlsUnset> {
806    client_builder: Option<hyper_util::client::legacy::Builder>,
807    pool_idle_timeout: Option<Option<Duration>>,
808    pool_max_idle_per_host: Option<usize>,
809    #[allow(unused)]
810    tls_provider: Tls,
811}
812
813cfg_tls! {
814    use aws_smithy_runtime_api::client::dns::ResolveDns;
815
816    impl ConnectorBuilder<TlsProviderSelected> {
817        /// Build a [`Connector`] that will use the default DNS resolver implementation.
818        pub fn build(self) -> Connector {
819            let http_connector = self.base_connector();
820            self.build_https(http_connector)
821        }
822
823        /// Configure the TLS context
824        pub fn tls_context(mut self, ctx: TlsContext) -> Self {
825            self.tls.context = ctx;
826            self
827        }
828
829        /// Configure the TLS context
830        pub fn set_tls_context(&mut self, ctx: TlsContext) -> &mut Self {
831            self.tls.context = ctx;
832            self
833        }
834
835        /// Build a [`Connector`] that will use the given DNS resolver implementation.
836        pub fn build_with_resolver<R: ResolveDns + Clone + 'static>(self, resolver: R) -> Connector {
837            use crate::client::dns::HyperUtilResolver;
838            let http_connector = self.base_connector_with_resolver(HyperUtilResolver { resolver });
839            self.build_https(http_connector)
840        }
841
842        fn build_https<R>(self, http_connector: HyperHttpConnector<R>) -> Connector
843        where
844            R: Clone + Send + Sync + 'static,
845            R: tower::Service<hyper_util::client::legacy::connect::dns::Name>,
846            R::Response: Iterator<Item = std::net::SocketAddr>,
847            R::Future: Send,
848            R::Error: Into<Box<dyn Error + Send + Sync>>,
849        {
850            match &self.tls.provider {
851                // TODO(hyper1) - fix cfg_rustls! to allow matching on patterns so we can re-use it and not duplicate these cfg matches everywhere
852                #[cfg(feature = "__rustls")]
853                tls::Provider::Rustls(crypto_mode) => {
854                    let proxy_config = self.proxy_config.clone()
855                        .unwrap_or_else(proxy::ProxyConfig::disabled);
856
857                    let https_connector = tls::rustls_provider::build_connector::wrap_connector(
858                        http_connector,
859                        crypto_mode.clone(),
860                        &self.tls.context,
861                        proxy_config,
862                    );
863                    self.wrap_connector(https_connector)
864                },
865                #[cfg(feature = "s2n-tls")]
866                tls::Provider::S2nTls  => {
867                    let proxy_config = self.proxy_config.clone()
868                        .unwrap_or_else(proxy::ProxyConfig::disabled);
869
870                    let https_connector = tls::s2n_tls_provider::build_connector::wrap_connector(
871                        http_connector,
872                        &self.tls.context,
873                        proxy_config,
874                    );
875                    self.wrap_connector(https_connector)
876                }
877            }
878        }
879    }
880
881    impl Builder<TlsProviderSelected> {
882        /// Create an HTTPS client with the selected TLS provider.
883        ///
884        /// The trusted certificates will be loaded later when this becomes the selected
885        /// HTTP client for a Smithy client.
886        pub fn build_https(self) -> SharedHttpClient {
887            build_with_conn_fn(
888                self.client_builder,
889                self.pool_idle_timeout,
890                self.pool_max_idle_per_host,
891                move |client_builder, settings, runtime_components| {
892                    let builder = new_conn_builder(client_builder, settings, runtime_components)
893                        .tls_provider(self.tls_provider.provider.clone())
894                        .tls_context(self.tls_provider.context.clone());
895                    builder.build()
896                },
897            )
898        }
899
900        /// Create an HTTPS client using a custom DNS resolver
901        pub fn build_with_resolver(
902            self,
903            resolver: impl ResolveDns + Clone + 'static,
904        ) -> SharedHttpClient {
905            build_with_conn_fn(
906                self.client_builder,
907                self.pool_idle_timeout,
908                self.pool_max_idle_per_host,
909                move |client_builder, settings, runtime_components| {
910                    let builder = new_conn_builder(client_builder, settings, runtime_components)
911                        .tls_provider(self.tls_provider.provider.clone())
912                        .tls_context(self.tls_provider.context.clone());
913                    builder.build_with_resolver(resolver.clone())
914                },
915            )
916        }
917
918        /// Configure the TLS context
919        pub fn tls_context(mut self, ctx: TlsContext) -> Self {
920            self.tls_provider.context = ctx;
921            self
922        }
923    }
924}
925
926impl<Any> Builder<Any> {
927    /// Set an optional timeout for idle sockets being kept-alive.
928    ///
929    /// Pass `None` to disable timeout.
930    ///
931    /// Defaults to Hyper's default timeout, which is currently 90 seconds - see
932    /// [hyper_util::client::legacy::Builder::pool_idle_timeout],
933    /// but unlike that function, there is no need to call `pool_timer` yourself.
934    ///
935    /// # Examples
936    ///
937    /// ```rust
938    /// # #[cfg(feature = "rustls-aws-lc")]
939    /// # {
940    /// use aws_smithy_http_client::{Builder, tls};
941    /// use std::time::Duration;
942    ///
943    /// let client = Builder::new()
944    ///     .pool_idle_timeout(Duration::from_secs(30))
945    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
946    ///     .build_https();
947    /// # }
948    /// # Ok::<(), Box<dyn std::error::Error>>(())
949    /// ```
950    pub fn pool_idle_timeout<D>(mut self, val: D) -> Self
951    where
952        D: Into<Option<Duration>>,
953    {
954        self.pool_idle_timeout = Some(val.into());
955        self
956    }
957
958    /// Set an optional timeout for idle sockets being kept-alive.
959    ///
960    /// Pass `None` to use Hyper's default timeout, `Some(None)` to disable timeouts.
961    ///
962    /// This is the mutable version of [`pool_idle_timeout`](Self::pool_idle_timeout).
963    ///
964    /// # Examples
965    ///
966    /// ```rust
967    /// # #[cfg(feature = "rustls-aws-lc")]
968    /// # {
969    /// use std::time::Duration;
970    /// use aws_smithy_http_client::{Builder, tls};
971    ///
972    /// let mut client = Builder::new();
973    /// client.set_pool_idle_timeout(Some(Some(Duration::from_secs(30))));
974    /// client
975    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
976    ///     .build_https();
977    /// # }
978    /// # Ok::<(), Box<dyn std::error::Error>>(())
979    /// ```
980    pub fn set_pool_idle_timeout(&mut self, val: Option<Option<Duration>>) -> &mut Self {
981        self.pool_idle_timeout = val;
982        self
983    }
984
985    /// Sets the maximum number of idle pooled connections allowed per host.
986    ///
987    /// Default is determined by the underlying hyper client, which is currently no limit
988    /// (`usize::MAX`) - see [hyper_util::client::legacy::Builder::pool_max_idle_per_host].
989    ///
990    /// # Examples
991    ///
992    /// ```rust
993    /// # #[cfg(feature = "rustls-aws-lc")]
994    /// # {
995    /// use aws_smithy_http_client::{Builder, tls};
996    ///
997    /// let client = Builder::new()
998    ///     .pool_max_idle_per_host(70)
999    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
1000    ///     .build_https();
1001    /// # }
1002    /// # Ok::<(), Box<dyn std::error::Error>>(())
1003    /// ```
1004    pub fn pool_max_idle_per_host(mut self, val: usize) -> Self {
1005        self.pool_max_idle_per_host = Some(val);
1006        self
1007    }
1008
1009    /// Sets the maximum number of idle pooled connections allowed per host.
1010    ///
1011    /// Pass `None` to use the hyper default (currently no limit, `usize::MAX`) - see
1012    /// [hyper_util::client::legacy::Builder::pool_max_idle_per_host].
1013    ///
1014    /// This is the mutable version of [`pool_max_idle_per_host`](Self::pool_max_idle_per_host).
1015    pub fn set_pool_max_idle_per_host(&mut self, val: Option<usize>) -> &mut Self {
1016        self.pool_max_idle_per_host = val;
1017        self
1018    }
1019}
1020
1021impl Builder<TlsUnset> {
1022    /// Creates a new builder.
1023    pub fn new() -> Self {
1024        Self::default()
1025    }
1026
1027    /// Returns a [`SharedHttpClient`] that calls the given `connector` function to select an HTTP(S) connector.
1028    #[doc(hidden)]
1029    pub fn build_with_connector_fn<F>(self, connector_fn: F) -> SharedHttpClient
1030    where
1031        F: Fn(Option<&HttpConnectorSettings>, Option<&RuntimeComponents>) -> Connector
1032            + Send
1033            + Sync
1034            + 'static,
1035    {
1036        build_with_conn_fn(
1037            self.client_builder,
1038            self.pool_idle_timeout,
1039            self.pool_max_idle_per_host,
1040            move |_builder, settings, runtime_components| {
1041                connector_fn(settings, runtime_components)
1042            },
1043        )
1044    }
1045
1046    /// Build a new HTTP client without TLS enabled
1047    #[doc(hidden)]
1048    pub fn build_http(self) -> SharedHttpClient {
1049        build_with_conn_fn(
1050            self.client_builder,
1051            self.pool_idle_timeout,
1052            self.pool_max_idle_per_host,
1053            move |client_builder, settings, runtime_components| {
1054                let builder = new_conn_builder(client_builder, settings, runtime_components);
1055                builder.build_http()
1056            },
1057        )
1058    }
1059
1060    /// Set the TLS implementation to use
1061    pub fn tls_provider(self, provider: tls::Provider) -> Builder<TlsProviderSelected> {
1062        Builder {
1063            client_builder: self.client_builder,
1064            pool_idle_timeout: self.pool_idle_timeout,
1065            pool_max_idle_per_host: self.pool_max_idle_per_host,
1066            tls_provider: TlsProviderSelected {
1067                provider,
1068                context: TlsContext::default(),
1069            },
1070        }
1071    }
1072}
1073
1074pub(crate) fn build_with_conn_fn<F>(
1075    client_builder: Option<hyper_util::client::legacy::Builder>,
1076    pool_idle_timeout: Option<Option<Duration>>,
1077    pool_max_idle_per_host: Option<usize>,
1078    connector_fn: F,
1079) -> SharedHttpClient
1080where
1081    F: Fn(
1082            hyper_util::client::legacy::Builder,
1083            Option<&HttpConnectorSettings>,
1084            Option<&RuntimeComponents>,
1085        ) -> Connector
1086        + Send
1087        + Sync
1088        + 'static,
1089{
1090    let client_builder = client_builder
1091        .unwrap_or_else(|| new_tokio_hyper_builder(pool_idle_timeout, pool_max_idle_per_host));
1092    SharedHttpClient::new(HyperClient {
1093        connector_cache: RwLock::new(HashMap::new()),
1094        client_builder,
1095        connector_fn,
1096    })
1097}
1098
1099#[allow(dead_code)]
1100pub(crate) fn build_with_tcp_conn_fn<C, F>(
1101    client_builder: Option<hyper_util::client::legacy::Builder>,
1102    pool_idle_timeout: Option<Option<Duration>>,
1103    pool_max_idle_per_host: Option<usize>,
1104    tcp_connector_fn: F,
1105) -> SharedHttpClient
1106where
1107    F: Fn() -> C + Send + Sync + 'static,
1108    C: Clone + Send + Sync + 'static,
1109    C: tower::Service<Uri>,
1110    C::Response: Connection + Read + Write + Send + Sync + Unpin + 'static,
1111    C::Future: Unpin + Send + 'static,
1112    C::Error: Into<BoxError>,
1113    C: Connect,
1114{
1115    build_with_conn_fn(
1116        client_builder,
1117        pool_idle_timeout,
1118        pool_max_idle_per_host,
1119        move |client_builder, settings, runtime_components| {
1120            let builder = new_conn_builder(client_builder, settings, runtime_components);
1121            builder.wrap_connector(tcp_connector_fn())
1122        },
1123    )
1124}
1125
1126fn new_conn_builder(
1127    client_builder: hyper_util::client::legacy::Builder,
1128    settings: Option<&HttpConnectorSettings>,
1129    runtime_components: Option<&RuntimeComponents>,
1130) -> ConnectorBuilder {
1131    let mut builder = Connector::builder().hyper_builder(client_builder);
1132    builder.set_connector_settings(settings.cloned());
1133    if let Some(components) = runtime_components {
1134        builder.set_sleep_impl(components.sleep_impl());
1135    }
1136    builder
1137}
1138
1139#[cfg(test)]
1140mod test {
1141    use std::io::{Error, ErrorKind};
1142    use std::pin::Pin;
1143    use std::sync::atomic::{AtomicU32, Ordering};
1144    use std::sync::Arc;
1145    use std::task::{Context, Poll};
1146
1147    use crate::client::timeout::test::NeverConnects;
1148    use aws_smithy_async::assert_elapsed;
1149    use aws_smithy_async::rt::sleep::TokioSleep;
1150    use aws_smithy_async::time::SystemTimeSource;
1151    use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
1152    use http_1x::Uri;
1153    use hyper::rt::ReadBufCursor;
1154    use hyper_util::client::legacy::connect::Connected;
1155
1156    use super::*;
1157
1158    #[tokio::test]
1159    async fn connector_selection() {
1160        // Create a client that increments a count every time it creates a new Connector
1161        let creation_count = Arc::new(AtomicU32::new(0));
1162        let http_client = build_with_tcp_conn_fn(None, None, None, {
1163            let count = creation_count.clone();
1164            move || {
1165                count.fetch_add(1, Ordering::Relaxed);
1166                NeverConnects
1167            }
1168        });
1169
1170        // This configuration should result in 4 separate connectors with different timeout settings
1171        let settings = [
1172            HttpConnectorSettings::builder()
1173                .connect_timeout(Duration::from_secs(3))
1174                .build(),
1175            HttpConnectorSettings::builder()
1176                .read_timeout(Duration::from_secs(3))
1177                .build(),
1178            HttpConnectorSettings::builder()
1179                .connect_timeout(Duration::from_secs(3))
1180                .read_timeout(Duration::from_secs(3))
1181                .build(),
1182            HttpConnectorSettings::builder()
1183                .connect_timeout(Duration::from_secs(5))
1184                .read_timeout(Duration::from_secs(3))
1185                .build(),
1186        ];
1187
1188        // Kick off thousands of parallel tasks that will try to create a connector
1189        let components = RuntimeComponentsBuilder::for_tests()
1190            .with_time_source(Some(SystemTimeSource::new()))
1191            .build()
1192            .unwrap();
1193        let mut handles = Vec::new();
1194        for setting in &settings {
1195            for _ in 0..1000 {
1196                let client = http_client.clone();
1197                handles.push(tokio::spawn({
1198                    let setting = setting.clone();
1199                    let components = components.clone();
1200                    async move {
1201                        let _ = client.http_connector(&setting, &components);
1202                    }
1203                }));
1204            }
1205        }
1206        for handle in handles {
1207            handle.await.unwrap();
1208        }
1209
1210        // Verify only 4 connectors were created amidst the chaos
1211        assert_eq!(4, creation_count.load(Ordering::Relaxed));
1212    }
1213
1214    #[tokio::test]
1215    async fn hyper_io_error() {
1216        let connector = TestConnection {
1217            inner: HangupStream,
1218        };
1219        let adapter = Connector::builder().wrap_connector(connector).adapter;
1220        let err = adapter
1221            .call(HttpRequest::get("https://socket-hangup.com").unwrap())
1222            .await
1223            .expect_err("socket hangup");
1224        assert!(err.is_io(), "unexpected error type: {:?}", err);
1225    }
1226
1227    // ---- machinery to make a Hyper connector that responds with an IO Error
1228    #[derive(Clone)]
1229    struct HangupStream;
1230
1231    impl Connection for HangupStream {
1232        fn connected(&self) -> Connected {
1233            Connected::new()
1234        }
1235    }
1236
1237    impl Read for HangupStream {
1238        fn poll_read(
1239            self: Pin<&mut Self>,
1240            _cx: &mut Context<'_>,
1241            _buf: ReadBufCursor<'_>,
1242        ) -> Poll<std::io::Result<()>> {
1243            Poll::Ready(Err(Error::new(
1244                ErrorKind::ConnectionReset,
1245                "connection reset",
1246            )))
1247        }
1248    }
1249
1250    impl Write for HangupStream {
1251        fn poll_write(
1252            self: Pin<&mut Self>,
1253            _cx: &mut Context<'_>,
1254            _buf: &[u8],
1255        ) -> Poll<Result<usize, Error>> {
1256            Poll::Pending
1257        }
1258
1259        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1260            Poll::Pending
1261        }
1262
1263        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1264            Poll::Pending
1265        }
1266    }
1267
1268    #[derive(Clone)]
1269    struct TestConnection<T> {
1270        inner: T,
1271    }
1272
1273    impl<T> tower::Service<Uri> for TestConnection<T>
1274    where
1275        T: Clone + Connection,
1276    {
1277        type Response = T;
1278        type Error = BoxError;
1279        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
1280
1281        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
1282            Poll::Ready(Ok(()))
1283        }
1284
1285        fn call(&mut self, _req: Uri) -> Self::Future {
1286            std::future::ready(Ok(self.inner.clone()))
1287        }
1288    }
1289
1290    #[tokio::test]
1291    async fn http_connect_timeout_works() {
1292        let tcp_connector = NeverConnects::default();
1293        let connector_settings = HttpConnectorSettings::builder()
1294            .connect_timeout(Duration::from_secs(1))
1295            .build();
1296        let hyper = Connector::builder()
1297            .connector_settings(connector_settings)
1298            .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1299            .wrap_connector(tcp_connector)
1300            .adapter;
1301        let now = tokio::time::Instant::now();
1302        tokio::time::pause();
1303        let resp = hyper
1304            .call(HttpRequest::get("https://static-uri.com").unwrap())
1305            .await
1306            .unwrap_err();
1307        assert!(
1308            resp.is_timeout(),
1309            "expected resp.is_timeout() to be true but it was false, resp == {:?}",
1310            resp
1311        );
1312        let message = DisplayErrorContext(&resp).to_string();
1313        let expected = "timeout: client error (Connect): HTTP connect timeout occurred after 1s";
1314        assert!(
1315            message.contains(expected),
1316            "expected '{message}' to contain '{expected}'"
1317        );
1318        assert_elapsed!(now, Duration::from_secs(1));
1319    }
1320
1321    #[tokio::test]
1322    async fn http_read_timeout_works() {
1323        let tcp_connector = crate::client::timeout::test::NeverReplies;
1324        let connector_settings = HttpConnectorSettings::builder()
1325            .connect_timeout(Duration::from_secs(1))
1326            .read_timeout(Duration::from_secs(2))
1327            .build();
1328        let hyper = Connector::builder()
1329            .connector_settings(connector_settings)
1330            .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1331            .wrap_connector(tcp_connector)
1332            .adapter;
1333        let now = tokio::time::Instant::now();
1334        tokio::time::pause();
1335        let err = hyper
1336            .call(HttpRequest::get("https://fake-uri.com").unwrap())
1337            .await
1338            .unwrap_err();
1339        assert!(
1340            err.is_timeout(),
1341            "expected err.is_timeout() to be true but it was false, err == {err:?}",
1342        );
1343        let message = format!("{}", DisplayErrorContext(&err));
1344        let expected = "timeout: HTTP read timeout occurred after 2s";
1345        assert!(
1346            message.contains(expected),
1347            "expected '{message}' to contain '{expected}'"
1348        );
1349        assert_elapsed!(now, Duration::from_secs(2));
1350    }
1351
1352    #[cfg(not(windows))]
1353    #[tokio::test]
1354    async fn connection_refused_works() {
1355        use crate::client::dns::HyperUtilResolver;
1356        use aws_smithy_runtime_api::client::dns::{DnsFuture, ResolveDns};
1357        use std::net::{IpAddr, Ipv4Addr};
1358
1359        #[derive(Debug, Clone, Default)]
1360        struct TestResolver;
1361        impl ResolveDns for TestResolver {
1362            fn resolve_dns<'a>(&'a self, _name: &'a str) -> DnsFuture<'a> {
1363                let localhost_v4 = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
1364                DnsFuture::ready(Ok(vec![localhost_v4]))
1365            }
1366        }
1367
1368        let connector_settings = HttpConnectorSettings::builder()
1369            .connect_timeout(Duration::from_secs(20))
1370            .build();
1371
1372        let resolver = HyperUtilResolver {
1373            resolver: TestResolver,
1374        };
1375        let connector = Connector::builder().base_connector_with_resolver(resolver);
1376
1377        let hyper = Connector::builder()
1378            .connector_settings(connector_settings)
1379            .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1380            .wrap_connector(connector)
1381            .adapter;
1382
1383        let resp = hyper
1384            .call(HttpRequest::get("http://static-uri:50227.com").unwrap())
1385            .await
1386            .unwrap_err();
1387        assert!(
1388            resp.is_io(),
1389            "expected resp.is_io() to be true but it was false, resp == {:?}",
1390            resp
1391        );
1392        let message = DisplayErrorContext(&resp).to_string();
1393        let expected = "Connection refused";
1394        assert!(
1395            message.contains(expected),
1396            "expected '{message}' to contain '{expected}'"
1397        );
1398    }
1399
1400    #[cfg(feature = "s2n-tls")]
1401    #[tokio::test]
1402    async fn s2n_tls_provider() {
1403        // Create an HttpConnector with the s2n-tls provider.
1404        let client = Builder::new()
1405            .tls_provider(tls::Provider::S2nTls)
1406            .build_https();
1407        let connector_settings = HttpConnectorSettings::builder().build();
1408
1409        // HyperClient::http_connector invokes TimeSource::now to determine how long it takes to
1410        // create new HttpConnectors. As such, a real time source must be provided.
1411        let runtime_components = RuntimeComponentsBuilder::for_tests()
1412            .with_time_source(Some(SystemTimeSource::new()))
1413            .build()
1414            .unwrap();
1415
1416        let connector = client.http_connector(&connector_settings, &runtime_components);
1417
1418        // Ensure that s2n-tls is used as the underlying TLS provider when selected.
1419        //
1420        // s2n-tls-hyper will error when given an invalid scheme. Ensure that this error is produced
1421        // from s2n-tls-hyper, and not another TLS provider.
1422        let error = connector
1423            .call(HttpRequest::get("notascheme://amazon.com").unwrap())
1424            .await
1425            .unwrap_err();
1426        let error = error.into_source();
1427        let s2n_error = error
1428            .source()
1429            .unwrap()
1430            .downcast_ref::<s2n_tls_hyper::error::Error>()
1431            .unwrap();
1432        assert!(matches!(
1433            s2n_error,
1434            s2n_tls_hyper::error::Error::InvalidScheme
1435        ));
1436    }
1437
1438    /// The default HTTP client reports "hyper" / "1.x" as its connector metadata.
1439    #[test]
1440    fn connector_metadata_identifies_hyper_1x() {
1441        let client = crate::Builder::new().build_http();
1442        let metadata = client
1443            .connector_metadata()
1444            .expect("connector metadata should be present");
1445        assert_eq!(metadata.name(), Cow::Borrowed("hyper"));
1446        assert_eq!(metadata.version(), Some(Cow::Borrowed("1.x")));
1447    }
1448}