1mod dns;
7pub mod pool;
8pub mod proxy;
10mod timeout;
11pub 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
60pub(crate) struct ConnectorCallTimer {
62 capture: CaptureHttpAttemptTelemetry,
63 time_source: SharedTimeSource,
64 started_at: SystemTime,
65}
66
67impl ConnectorCallTimer {
68 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 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
89pub 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#[derive(Debug)]
124pub struct Connector {
125 adapter: Box<dyn HttpConnector>,
126}
127
128impl Connector {
129 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#[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 enable_tcp_nodelay: true,
172 interface: None,
173 proxy_config: None,
174 tls: Tls::default(),
175 }
176 }
177}
178
179#[derive(Default, Debug, Clone)]
181#[non_exhaustive]
182pub struct TlsUnset {}
183
184#[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 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 #[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 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 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 fn base_connector(&self) -> HyperHttpConnector {
301 self.base_connector_with_resolver(GaiResolver::new())
302 }
303
304 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 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 pub fn set_sleep_impl(&mut self, sleep_impl: Option<SharedAsyncSleep>) -> &mut Self {
330 self.sleep_impl = sleep_impl;
331 self
332 }
333
334 fn set_time_source(&mut self, time_source: SharedTimeSource) -> &mut Self {
336 self.time_source = time_source;
337 self
338 }
339
340 pub fn connector_settings(mut self, connector_settings: HttpConnectorSettings) -> Self {
342 self.connector_settings = Some(connector_settings);
343 self
344 }
345
346 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 pub fn enable_tcp_nodelay(mut self, nodelay: bool) -> Self {
357 self.enable_tcp_nodelay = nodelay;
358 self
359 }
360
361 pub fn set_enable_tcp_nodelay(&mut self, nodelay: bool) -> &mut Self {
363 self.enable_tcp_nodelay = nodelay;
364 self
365 }
366
367 #[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 pub fn proxy_config(mut self, config: proxy::ProxyConfig) -> Self {
407 self.proxy_config = Some(config);
408 self
409 }
410
411 pub fn set_proxy_config(&mut self, config: Option<proxy::ProxyConfig>) -> &mut Self {
415 self.proxy_config = config;
416 self
417 }
418
419 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 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 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 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 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 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
536struct 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
556fn 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 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 fn add_proxy_auth_header(&self, request: &mut http_1x::Request<SdkBody>) {
605 if request.uri().scheme() != Some(&http_1x::uri::Scheme::HTTP) {
607 return;
608 }
609
610 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 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
676fn downcast_error(err: BoxError) -> ConnectorError {
678 if find_source::<TimedOutError>(err.as_ref()).is_some() {
680 return ConnectorError::timeout(err);
681 }
682 let err = match err.downcast::<ConnectorError>() {
684 Ok(connector_error) => return *connector_error,
685 Err(box_error) => box_error,
686 };
687 if find_source::<ResolveDnsError>(err.as_ref()).is_some() {
689 return ConnectorError::io(err);
690 }
691 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 ConnectorError::other(err, None)
710}
711
712fn 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 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#[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 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 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#[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 pub fn build(self) -> Connector {
871 let http_connector = self.base_connector();
872 self.build_https(http_connector)
873 }
874
875 pub fn tls_context(mut self, ctx: TlsContext) -> Self {
877 self.tls.context = ctx;
878 self
879 }
880
881 pub fn set_tls_context(&mut self, ctx: TlsContext) -> &mut Self {
883 self.tls.context = ctx;
884 self
885 }
886
887 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 #[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 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 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 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 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 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 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 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 pub fn new() -> Self {
1076 Self::default()
1077 }
1078
1079 #[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 #[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 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 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 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 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 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 #[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 let client = Builder::new()
1487 .tls_provider(tls::Provider::S2nTls)
1488 .build_https();
1489 let connector_settings = HttpConnectorSettings::builder().build();
1490
1491 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 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 #[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}