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// TODO(https://github.com/awslabs/aws-sdk-rust/issues/1090): CacheKey must also include ptr equality to any
752// runtime components that are used—sleep_impl as a base (unless we prohibit overriding sleep impl)
753// If we decide to put a DnsResolver in RuntimeComponents, then we'll need to handle that as well.
754#[derive(Clone, Debug, Eq, PartialEq, Hash)]
755struct CacheKey {
756    connect_timeout: Option<Duration>,
757    read_timeout: Option<Duration>,
758}
759
760impl From<&HttpConnectorSettings> for CacheKey {
761    fn from(value: &HttpConnectorSettings) -> Self {
762        Self {
763            connect_timeout: value.connect_timeout(),
764            read_timeout: value.read_timeout(),
765        }
766    }
767}
768
769struct HyperClient<F> {
770    connector_cache: RwLock<HashMap<CacheKey, SharedHttpConnector>>,
771    client_builder: hyper_util::client::legacy::Builder,
772    connector_fn: F,
773}
774
775impl<F> fmt::Debug for HyperClient<F> {
776    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
777        f.debug_struct("HyperClient")
778            .field("connector_cache", &self.connector_cache)
779            .field("client_builder", &self.client_builder)
780            .finish()
781    }
782}
783
784impl<F> HttpClient for HyperClient<F>
785where
786    F: Fn(
787            hyper_util::client::legacy::Builder,
788            Option<&HttpConnectorSettings>,
789            Option<&RuntimeComponents>,
790        ) -> Connector
791        + Send
792        + Sync
793        + 'static,
794{
795    fn http_connector(
796        &self,
797        settings: &HttpConnectorSettings,
798        components: &RuntimeComponents,
799    ) -> SharedHttpConnector {
800        let key = CacheKey::from(settings);
801        let mut connector = self.connector_cache.read().unwrap().get(&key).cloned();
802        if connector.is_none() {
803            let mut cache = self.connector_cache.write().unwrap();
804            // Short-circuit if another thread already wrote a connector to the cache for this key
805            if !cache.contains_key(&key) {
806                let start = components.time_source().map(|ts| ts.now());
807                let connector = (self.connector_fn)(
808                    self.client_builder.clone(),
809                    Some(settings),
810                    Some(components),
811                );
812                let end = components.time_source().map(|ts| ts.now());
813                if let (Some(start), Some(end)) = (start, end) {
814                    if let Ok(elapsed) = end.duration_since(start) {
815                        tracing::debug!("new connector created in {:?}", elapsed);
816                    }
817                }
818                let connector = SharedHttpConnector::new(connector);
819                cache.insert(key.clone(), connector);
820            }
821            connector = cache.get(&key).cloned();
822        }
823
824        connector.expect("cache populated above")
825    }
826
827    fn validate_base_client_config(
828        &self,
829        _: &RuntimeComponentsBuilder,
830        _: &ConfigBag,
831    ) -> Result<(), BoxError> {
832        // Initialize the TCP connector at this point so that native certs load
833        // at client initialization time instead of upon first request. We do it
834        // here rather than at construction so that it won't run if this is not
835        // the selected HTTP client for the base config (for example, if this was
836        // the default HTTP client, and it was overridden by a later plugin).
837        let _ = (self.connector_fn)(self.client_builder.clone(), None, None);
838        Ok(())
839    }
840
841    fn connector_metadata(&self) -> Option<ConnectorMetadata> {
842        Some(ConnectorMetadata::new("hyper", Some(Cow::Borrowed("1.x"))))
843    }
844}
845
846/// Builder for a hyper-backed [`HttpClient`] implementation.
847///
848/// This builder can be used to customize the underlying TCP connector used, as well as
849/// hyper client configuration.
850///
851/// # Examples
852///
853/// Construct a Hyper client with the RusTLS TLS implementation.
854/// This can be useful when you want to share a Hyper connector between multiple
855/// generated Smithy clients.
856#[derive(Clone, Default, Debug)]
857pub struct Builder<Tls = TlsUnset> {
858    client_builder: Option<hyper_util::client::legacy::Builder>,
859    pool_idle_timeout: Option<Option<Duration>>,
860    pool_max_idle_per_host: Option<usize>,
861    #[allow(unused)]
862    tls_provider: Tls,
863}
864
865cfg_tls! {
866    use aws_smithy_runtime_api::client::dns::ResolveDns;
867
868    impl ConnectorBuilder<TlsProviderSelected> {
869        /// Build a [`Connector`] that will use the default DNS resolver implementation.
870        pub fn build(self) -> Connector {
871            let http_connector = self.base_connector();
872            self.build_https(http_connector)
873        }
874
875        /// Configure the TLS context
876        pub fn tls_context(mut self, ctx: TlsContext) -> Self {
877            self.tls.context = ctx;
878            self
879        }
880
881        /// Configure the TLS context
882        pub fn set_tls_context(&mut self, ctx: TlsContext) -> &mut Self {
883            self.tls.context = ctx;
884            self
885        }
886
887        /// Build a [`Connector`] that will use the given DNS resolver implementation.
888        pub fn build_with_resolver<R: ResolveDns + Clone + 'static>(self, resolver: R) -> Connector {
889            use crate::client::dns::HyperUtilResolver;
890            let http_connector = self.base_connector_with_resolver(HyperUtilResolver { resolver });
891            self.build_https(http_connector)
892        }
893
894        fn build_https<R>(self, http_connector: HyperHttpConnector<R>) -> Connector
895        where
896            R: Clone + Send + Sync + 'static,
897            R: tower::Service<hyper_util::client::legacy::connect::dns::Name>,
898            R::Response: Iterator<Item = std::net::SocketAddr>,
899            R::Future: Send,
900            R::Error: Into<Box<dyn Error + Send + Sync>>,
901        {
902            match &self.tls.provider {
903                // TODO(hyper1) - fix cfg_rustls! to allow matching on patterns so we can re-use it and not duplicate these cfg matches everywhere
904                #[cfg(feature = "__rustls")]
905                tls::Provider::Rustls(crypto_mode) => {
906                    let proxy_config = self.proxy_config.clone()
907                        .unwrap_or_else(proxy::ProxyConfig::disabled);
908
909                    let https_connector = tls::rustls_provider::build_connector::wrap_connector(
910                        http_connector,
911                        crypto_mode.clone(),
912                        &self.tls.context,
913                        proxy_config,
914                    );
915                    self.wrap_connector(https_connector)
916                },
917                #[cfg(feature = "s2n-tls")]
918                tls::Provider::S2nTls  => {
919                    let proxy_config = self.proxy_config.clone()
920                        .unwrap_or_else(proxy::ProxyConfig::disabled);
921
922                    let https_connector = tls::s2n_tls_provider::build_connector::wrap_connector(
923                        http_connector,
924                        &self.tls.context,
925                        proxy_config,
926                    );
927                    self.wrap_connector(https_connector)
928                }
929            }
930        }
931    }
932
933    impl Builder<TlsProviderSelected> {
934        /// Create an HTTPS client with the selected TLS provider.
935        ///
936        /// The trusted certificates will be loaded later when this becomes the selected
937        /// HTTP client for a Smithy client.
938        pub fn build_https(self) -> SharedHttpClient {
939            build_with_conn_fn(
940                self.client_builder,
941                self.pool_idle_timeout,
942                self.pool_max_idle_per_host,
943                move |client_builder, settings, runtime_components| {
944                    let builder = new_conn_builder(client_builder, settings, runtime_components)
945                        .tls_provider(self.tls_provider.provider.clone())
946                        .tls_context(self.tls_provider.context.clone());
947                    builder.build()
948                },
949            )
950        }
951
952        /// Create an HTTPS client using a custom DNS resolver
953        pub fn build_with_resolver(
954            self,
955            resolver: impl ResolveDns + Clone + 'static,
956        ) -> SharedHttpClient {
957            build_with_conn_fn(
958                self.client_builder,
959                self.pool_idle_timeout,
960                self.pool_max_idle_per_host,
961                move |client_builder, settings, runtime_components| {
962                    let builder = new_conn_builder(client_builder, settings, runtime_components)
963                        .tls_provider(self.tls_provider.provider.clone())
964                        .tls_context(self.tls_provider.context.clone());
965                    builder.build_with_resolver(resolver.clone())
966                },
967            )
968        }
969
970        /// Configure the TLS context
971        pub fn tls_context(mut self, ctx: TlsContext) -> Self {
972            self.tls_provider.context = ctx;
973            self
974        }
975    }
976}
977
978impl<Any> Builder<Any> {
979    /// Set an optional timeout for idle sockets being kept-alive.
980    ///
981    /// Pass `None` to disable timeout.
982    ///
983    /// Defaults to Hyper's default timeout, which is currently 90 seconds - see
984    /// [hyper_util::client::legacy::Builder::pool_idle_timeout],
985    /// but unlike that function, there is no need to call `pool_timer` yourself.
986    ///
987    /// # Examples
988    ///
989    /// ```rust
990    /// # #[cfg(feature = "rustls-aws-lc")]
991    /// # {
992    /// use aws_smithy_http_client::{Builder, tls};
993    /// use std::time::Duration;
994    ///
995    /// let client = Builder::new()
996    ///     .pool_idle_timeout(Duration::from_secs(30))
997    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
998    ///     .build_https();
999    /// # }
1000    /// # Ok::<(), Box<dyn std::error::Error>>(())
1001    /// ```
1002    pub fn pool_idle_timeout<D>(mut self, val: D) -> Self
1003    where
1004        D: Into<Option<Duration>>,
1005    {
1006        self.pool_idle_timeout = Some(val.into());
1007        self
1008    }
1009
1010    /// Set an optional timeout for idle sockets being kept-alive.
1011    ///
1012    /// Pass `None` to use Hyper's default timeout, `Some(None)` to disable timeouts.
1013    ///
1014    /// This is the mutable version of [`pool_idle_timeout`](Self::pool_idle_timeout).
1015    ///
1016    /// # Examples
1017    ///
1018    /// ```rust
1019    /// # #[cfg(feature = "rustls-aws-lc")]
1020    /// # {
1021    /// use std::time::Duration;
1022    /// use aws_smithy_http_client::{Builder, tls};
1023    ///
1024    /// let mut client = Builder::new();
1025    /// client.set_pool_idle_timeout(Some(Some(Duration::from_secs(30))));
1026    /// client
1027    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
1028    ///     .build_https();
1029    /// # }
1030    /// # Ok::<(), Box<dyn std::error::Error>>(())
1031    /// ```
1032    pub fn set_pool_idle_timeout(&mut self, val: Option<Option<Duration>>) -> &mut Self {
1033        self.pool_idle_timeout = val;
1034        self
1035    }
1036
1037    /// Sets the maximum number of idle pooled connections allowed per host.
1038    ///
1039    /// Default is determined by the underlying hyper client, which is currently no limit
1040    /// (`usize::MAX`) - see [hyper_util::client::legacy::Builder::pool_max_idle_per_host].
1041    ///
1042    /// # Examples
1043    ///
1044    /// ```rust
1045    /// # #[cfg(feature = "rustls-aws-lc")]
1046    /// # {
1047    /// use aws_smithy_http_client::{Builder, tls};
1048    ///
1049    /// let client = Builder::new()
1050    ///     .pool_max_idle_per_host(70)
1051    ///     .tls_provider(tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc))
1052    ///     .build_https();
1053    /// # }
1054    /// # Ok::<(), Box<dyn std::error::Error>>(())
1055    /// ```
1056    pub fn pool_max_idle_per_host(mut self, val: usize) -> Self {
1057        self.pool_max_idle_per_host = Some(val);
1058        self
1059    }
1060
1061    /// Sets the maximum number of idle pooled connections allowed per host.
1062    ///
1063    /// Pass `None` to use the hyper default (currently no limit, `usize::MAX`) - see
1064    /// [hyper_util::client::legacy::Builder::pool_max_idle_per_host].
1065    ///
1066    /// This is the mutable version of [`pool_max_idle_per_host`](Self::pool_max_idle_per_host).
1067    pub fn set_pool_max_idle_per_host(&mut self, val: Option<usize>) -> &mut Self {
1068        self.pool_max_idle_per_host = val;
1069        self
1070    }
1071}
1072
1073impl Builder<TlsUnset> {
1074    /// Creates a new builder.
1075    pub fn new() -> Self {
1076        Self::default()
1077    }
1078
1079    /// Returns a [`SharedHttpClient`] that calls the given `connector` function to select an HTTP(S) connector.
1080    #[doc(hidden)]
1081    pub fn build_with_connector_fn<F>(self, connector_fn: F) -> SharedHttpClient
1082    where
1083        F: Fn(Option<&HttpConnectorSettings>, Option<&RuntimeComponents>) -> Connector
1084            + Send
1085            + Sync
1086            + 'static,
1087    {
1088        build_with_conn_fn(
1089            self.client_builder,
1090            self.pool_idle_timeout,
1091            self.pool_max_idle_per_host,
1092            move |_builder, settings, runtime_components| {
1093                connector_fn(settings, runtime_components)
1094            },
1095        )
1096    }
1097
1098    /// Build a new HTTP client without TLS enabled
1099    #[doc(hidden)]
1100    pub fn build_http(self) -> SharedHttpClient {
1101        build_with_conn_fn(
1102            self.client_builder,
1103            self.pool_idle_timeout,
1104            self.pool_max_idle_per_host,
1105            move |client_builder, settings, runtime_components| {
1106                let builder = new_conn_builder(client_builder, settings, runtime_components);
1107                builder.build_http()
1108            },
1109        )
1110    }
1111
1112    /// Set the TLS implementation to use
1113    pub fn tls_provider(self, provider: tls::Provider) -> Builder<TlsProviderSelected> {
1114        Builder {
1115            client_builder: self.client_builder,
1116            pool_idle_timeout: self.pool_idle_timeout,
1117            pool_max_idle_per_host: self.pool_max_idle_per_host,
1118            tls_provider: TlsProviderSelected {
1119                provider,
1120                context: TlsContext::default(),
1121            },
1122        }
1123    }
1124}
1125
1126pub(crate) fn build_with_conn_fn<F>(
1127    client_builder: Option<hyper_util::client::legacy::Builder>,
1128    pool_idle_timeout: Option<Option<Duration>>,
1129    pool_max_idle_per_host: Option<usize>,
1130    connector_fn: F,
1131) -> SharedHttpClient
1132where
1133    F: Fn(
1134            hyper_util::client::legacy::Builder,
1135            Option<&HttpConnectorSettings>,
1136            Option<&RuntimeComponents>,
1137        ) -> Connector
1138        + Send
1139        + Sync
1140        + 'static,
1141{
1142    let client_builder = client_builder
1143        .unwrap_or_else(|| new_tokio_hyper_builder(pool_idle_timeout, pool_max_idle_per_host));
1144    SharedHttpClient::new(HyperClient {
1145        connector_cache: RwLock::new(HashMap::new()),
1146        client_builder,
1147        connector_fn,
1148    })
1149}
1150
1151#[allow(dead_code)]
1152pub(crate) fn build_with_tcp_conn_fn<C, F>(
1153    client_builder: Option<hyper_util::client::legacy::Builder>,
1154    pool_idle_timeout: Option<Option<Duration>>,
1155    pool_max_idle_per_host: Option<usize>,
1156    tcp_connector_fn: F,
1157) -> SharedHttpClient
1158where
1159    F: Fn() -> C + Send + Sync + 'static,
1160    C: Clone + Send + Sync + 'static,
1161    C: tower::Service<Uri>,
1162    C::Response: Connection + Read + Write + Send + Sync + Unpin + 'static,
1163    C::Future: Unpin + Send + 'static,
1164    C::Error: Into<BoxError>,
1165    C: Connect,
1166{
1167    build_with_conn_fn(
1168        client_builder,
1169        pool_idle_timeout,
1170        pool_max_idle_per_host,
1171        move |client_builder, settings, runtime_components| {
1172            let builder = new_conn_builder(client_builder, settings, runtime_components);
1173            builder.wrap_connector(tcp_connector_fn())
1174        },
1175    )
1176}
1177
1178fn new_conn_builder(
1179    client_builder: hyper_util::client::legacy::Builder,
1180    settings: Option<&HttpConnectorSettings>,
1181    runtime_components: Option<&RuntimeComponents>,
1182) -> ConnectorBuilder {
1183    let mut builder = Connector::builder().hyper_builder(client_builder);
1184    builder.set_connector_settings(settings.cloned());
1185    if let Some(components) = runtime_components {
1186        builder.set_sleep_impl(components.sleep_impl());
1187        if let Some(time_source) = components.time_source() {
1188            builder.set_time_source(time_source);
1189        }
1190    }
1191    builder
1192}
1193
1194#[cfg(test)]
1195mod test {
1196    use std::io::{Error, ErrorKind};
1197    use std::pin::Pin;
1198    use std::sync::atomic::{AtomicU32, Ordering};
1199    use std::sync::Arc;
1200    use std::task::{Context, Poll};
1201
1202    use crate::client::timeout::test::NeverConnects;
1203    use aws_smithy_async::assert_elapsed;
1204    use aws_smithy_async::rt::sleep::TokioSleep;
1205    use aws_smithy_async::test_util::ManualTimeSource;
1206    use aws_smithy_async::time::SystemTimeSource;
1207    use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
1208    use http_1x::Uri;
1209    use hyper::rt::ReadBufCursor;
1210    use hyper_util::client::legacy::connect::Connected;
1211    use std::time::UNIX_EPOCH;
1212
1213    #[test]
1214    fn connector_call_timer_ignores_a_backwards_clock() {
1215        let capture = CaptureHttpAttemptTelemetry::new();
1216        let time_source = SharedTimeSource::new(ManualTimeSource::new(SystemTime::UNIX_EPOCH));
1217
1218        ConnectorCallTimer {
1219            capture: capture.clone(),
1220            time_source,
1221            started_at: SystemTime::UNIX_EPOCH + Duration::from_secs(1),
1222        }
1223        .finish();
1224
1225        assert_eq!(None, capture.get().connector_call_duration());
1226    }
1227
1228    use super::*;
1229
1230    #[tokio::test]
1231    async fn connector_selection() {
1232        // Create a client that increments a count every time it creates a new Connector
1233        let creation_count = Arc::new(AtomicU32::new(0));
1234        let http_client = build_with_tcp_conn_fn(None, None, None, {
1235            let count = creation_count.clone();
1236            move || {
1237                count.fetch_add(1, Ordering::Relaxed);
1238                NeverConnects
1239            }
1240        });
1241
1242        // This configuration should result in 4 separate connectors with different timeout settings
1243        let settings = [
1244            HttpConnectorSettings::builder()
1245                .connect_timeout(Duration::from_secs(3))
1246                .build(),
1247            HttpConnectorSettings::builder()
1248                .read_timeout(Duration::from_secs(3))
1249                .build(),
1250            HttpConnectorSettings::builder()
1251                .connect_timeout(Duration::from_secs(3))
1252                .read_timeout(Duration::from_secs(3))
1253                .build(),
1254            HttpConnectorSettings::builder()
1255                .connect_timeout(Duration::from_secs(5))
1256                .read_timeout(Duration::from_secs(3))
1257                .build(),
1258        ];
1259
1260        // Kick off thousands of parallel tasks that will try to create a connector
1261        let components = RuntimeComponentsBuilder::for_tests()
1262            .with_time_source(Some(SystemTimeSource::new()))
1263            .build()
1264            .unwrap();
1265        let mut handles = Vec::new();
1266        for setting in &settings {
1267            for _ in 0..1000 {
1268                let client = http_client.clone();
1269                handles.push(tokio::spawn({
1270                    let setting = setting.clone();
1271                    let components = components.clone();
1272                    async move {
1273                        let _ = client.http_connector(&setting, &components);
1274                    }
1275                }));
1276            }
1277        }
1278        for handle in handles {
1279            handle.await.unwrap();
1280        }
1281
1282        // Verify only 4 connectors were created amidst the chaos
1283        assert_eq!(4, creation_count.load(Ordering::Relaxed));
1284    }
1285
1286    #[tokio::test]
1287    async fn hyper_io_error() {
1288        let time = ManualTimeSource::new(UNIX_EPOCH);
1289        let connector = TestConnection {
1290            inner: HangupStream { time: time.clone() },
1291        };
1292        let mut builder = Connector::builder();
1293        builder.set_time_source(SharedTimeSource::new(time));
1294        let adapter = builder.wrap_connector(connector).adapter;
1295        let capture = CaptureHttpAttemptTelemetry::new();
1296        let mut request = HttpRequest::get("https://socket-hangup.com").unwrap();
1297        request.add_extension(capture.clone());
1298        let err = adapter.call(request).await.expect_err("socket hangup");
1299        assert!(err.is_io(), "unexpected error type: {:?}", err);
1300        assert_eq!(
1301            capture.get().connector_call_duration(),
1302            Some(Duration::from_secs(7))
1303        );
1304    }
1305
1306    // ---- machinery to make a Hyper connector that responds with an IO Error
1307    #[derive(Clone)]
1308    struct HangupStream {
1309        time: ManualTimeSource,
1310    }
1311
1312    impl Connection for HangupStream {
1313        fn connected(&self) -> Connected {
1314            Connected::new()
1315        }
1316    }
1317
1318    impl Read for HangupStream {
1319        fn poll_read(
1320            self: Pin<&mut Self>,
1321            _cx: &mut Context<'_>,
1322            _buf: ReadBufCursor<'_>,
1323        ) -> Poll<std::io::Result<()>> {
1324            self.get_mut().time.advance(Duration::from_secs(7));
1325            Poll::Ready(Err(Error::new(
1326                ErrorKind::ConnectionReset,
1327                "connection reset",
1328            )))
1329        }
1330    }
1331
1332    impl Write for HangupStream {
1333        fn poll_write(
1334            self: Pin<&mut Self>,
1335            _cx: &mut Context<'_>,
1336            _buf: &[u8],
1337        ) -> Poll<Result<usize, Error>> {
1338            Poll::Pending
1339        }
1340
1341        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1342            Poll::Pending
1343        }
1344
1345        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1346            Poll::Pending
1347        }
1348    }
1349
1350    #[derive(Clone)]
1351    struct TestConnection<T> {
1352        inner: T,
1353    }
1354
1355    impl<T> tower::Service<Uri> for TestConnection<T>
1356    where
1357        T: Clone + Connection,
1358    {
1359        type Response = T;
1360        type Error = BoxError;
1361        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
1362
1363        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
1364            Poll::Ready(Ok(()))
1365        }
1366
1367        fn call(&mut self, _req: Uri) -> Self::Future {
1368            std::future::ready(Ok(self.inner.clone()))
1369        }
1370    }
1371
1372    #[tokio::test]
1373    async fn http_connect_timeout_works() {
1374        let tcp_connector = NeverConnects::default();
1375        let connector_settings = HttpConnectorSettings::builder()
1376            .connect_timeout(Duration::from_secs(1))
1377            .build();
1378        let hyper = Connector::builder()
1379            .connector_settings(connector_settings)
1380            .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1381            .wrap_connector(tcp_connector)
1382            .adapter;
1383        let now = tokio::time::Instant::now();
1384        tokio::time::pause();
1385        let resp = hyper
1386            .call(HttpRequest::get("https://static-uri.com").unwrap())
1387            .await
1388            .unwrap_err();
1389        assert!(
1390            resp.is_timeout(),
1391            "expected resp.is_timeout() to be true but it was false, resp == {:?}",
1392            resp
1393        );
1394        let message = DisplayErrorContext(&resp).to_string();
1395        let expected = "timeout: client error (Connect): HTTP connect timeout occurred after 1s";
1396        assert!(
1397            message.contains(expected),
1398            "expected '{message}' to contain '{expected}'"
1399        );
1400        assert_elapsed!(now, Duration::from_secs(1));
1401    }
1402
1403    #[tokio::test]
1404    async fn http_read_timeout_works() {
1405        let tcp_connector = crate::client::timeout::test::NeverReplies;
1406        let connector_settings = HttpConnectorSettings::builder()
1407            .connect_timeout(Duration::from_secs(1))
1408            .read_timeout(Duration::from_secs(2))
1409            .build();
1410        let hyper = Connector::builder()
1411            .connector_settings(connector_settings)
1412            .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1413            .wrap_connector(tcp_connector)
1414            .adapter;
1415        let now = tokio::time::Instant::now();
1416        tokio::time::pause();
1417        let err = hyper
1418            .call(HttpRequest::get("https://fake-uri.com").unwrap())
1419            .await
1420            .unwrap_err();
1421        assert!(
1422            err.is_timeout(),
1423            "expected err.is_timeout() to be true but it was false, err == {err:?}",
1424        );
1425        let message = format!("{}", DisplayErrorContext(&err));
1426        let expected = "timeout: HTTP read timeout occurred after 2s";
1427        assert!(
1428            message.contains(expected),
1429            "expected '{message}' to contain '{expected}'"
1430        );
1431        assert_elapsed!(now, Duration::from_secs(2));
1432    }
1433
1434    #[cfg(not(windows))]
1435    #[tokio::test]
1436    async fn connection_refused_works() {
1437        use crate::client::dns::HyperUtilResolver;
1438        use aws_smithy_runtime_api::client::dns::{DnsFuture, ResolveDns};
1439        use std::net::{IpAddr, Ipv4Addr};
1440
1441        #[derive(Debug, Clone, Default)]
1442        struct TestResolver;
1443        impl ResolveDns for TestResolver {
1444            fn resolve_dns<'a>(&'a self, _name: &'a str) -> DnsFuture<'a> {
1445                let localhost_v4 = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
1446                DnsFuture::ready(Ok(vec![localhost_v4]))
1447            }
1448        }
1449
1450        let connector_settings = HttpConnectorSettings::builder()
1451            .connect_timeout(Duration::from_secs(20))
1452            .build();
1453
1454        let resolver = HyperUtilResolver {
1455            resolver: TestResolver,
1456        };
1457        let connector = Connector::builder().base_connector_with_resolver(resolver);
1458
1459        let hyper = Connector::builder()
1460            .connector_settings(connector_settings)
1461            .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1462            .wrap_connector(connector)
1463            .adapter;
1464
1465        let resp = hyper
1466            .call(HttpRequest::get("http://static-uri:50227.com").unwrap())
1467            .await
1468            .unwrap_err();
1469        assert!(
1470            resp.is_io(),
1471            "expected resp.is_io() to be true but it was false, resp == {:?}",
1472            resp
1473        );
1474        let message = DisplayErrorContext(&resp).to_string();
1475        let expected = "Connection refused";
1476        assert!(
1477            message.contains(expected),
1478            "expected '{message}' to contain '{expected}'"
1479        );
1480    }
1481
1482    #[cfg(feature = "s2n-tls")]
1483    #[tokio::test]
1484    async fn s2n_tls_provider() {
1485        // Create an HttpConnector with the s2n-tls provider.
1486        let client = Builder::new()
1487            .tls_provider(tls::Provider::S2nTls)
1488            .build_https();
1489        let connector_settings = HttpConnectorSettings::builder().build();
1490
1491        // HyperClient::http_connector invokes TimeSource::now to determine how long it takes to
1492        // create new HttpConnectors. As such, a real time source must be provided.
1493        let runtime_components = RuntimeComponentsBuilder::for_tests()
1494            .with_time_source(Some(SystemTimeSource::new()))
1495            .build()
1496            .unwrap();
1497
1498        let connector = client.http_connector(&connector_settings, &runtime_components);
1499
1500        // Ensure that s2n-tls is used as the underlying TLS provider when selected.
1501        //
1502        // s2n-tls-hyper will error when given an invalid scheme. Ensure that this error is produced
1503        // from s2n-tls-hyper, and not another TLS provider.
1504        let error = connector
1505            .call(HttpRequest::get("notascheme://amazon.com").unwrap())
1506            .await
1507            .unwrap_err();
1508        let error = error.into_source();
1509        let s2n_error = find_source::<s2n_tls_hyper::error::Error>(error.as_ref())
1510            .expect("s2n-tls-hyper error is present in the source chain");
1511        assert!(matches!(
1512            s2n_error,
1513            s2n_tls_hyper::error::Error::InvalidScheme
1514        ));
1515    }
1516
1517    /// The default HTTP client reports "hyper" / "1.x" as its connector metadata.
1518    #[test]
1519    fn connector_metadata_identifies_hyper_1x() {
1520        let client = crate::Builder::new().build_http();
1521        let metadata = client
1522            .connector_metadata()
1523            .expect("connector metadata should be present");
1524        assert_eq!(metadata.name(), Cow::Borrowed("hyper"));
1525        assert_eq!(metadata.version(), Some(Cow::Borrowed("1.x")));
1526    }
1527}