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