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
751pub(crate) fn error_chain<'a>(
757 error: &'a (dyn Error + 'static),
758) -> impl Iterator<Item = &'a (dyn Error + 'static)> {
759 std::iter::successors(Some(error), |&error| {
760 match error.downcast_ref::<std::io::Error>() {
761 Some(io) => io.get_ref().map(|inner| inner as &(dyn Error + 'static)),
762 None => error.source(),
763 }
764 })
765}
766
767#[derive(Clone, Debug, Eq, PartialEq, Hash)]
771struct CacheKey {
772 connect_timeout: Option<Duration>,
773 read_timeout: Option<Duration>,
774}
775
776impl From<&HttpConnectorSettings> for CacheKey {
777 fn from(value: &HttpConnectorSettings) -> Self {
778 Self {
779 connect_timeout: value.connect_timeout(),
780 read_timeout: value.read_timeout(),
781 }
782 }
783}
784
785struct HyperClient<F> {
786 connector_cache: RwLock<HashMap<CacheKey, SharedHttpConnector>>,
787 client_builder: hyper_util::client::legacy::Builder,
788 connector_fn: F,
789}
790
791impl<F> fmt::Debug for HyperClient<F> {
792 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
793 f.debug_struct("HyperClient")
794 .field("connector_cache", &self.connector_cache)
795 .field("client_builder", &self.client_builder)
796 .finish()
797 }
798}
799
800impl<F> HttpClient for HyperClient<F>
801where
802 F: Fn(
803 hyper_util::client::legacy::Builder,
804 Option<&HttpConnectorSettings>,
805 Option<&RuntimeComponents>,
806 ) -> Connector
807 + Send
808 + Sync
809 + 'static,
810{
811 fn http_connector(
812 &self,
813 settings: &HttpConnectorSettings,
814 components: &RuntimeComponents,
815 ) -> SharedHttpConnector {
816 let key = CacheKey::from(settings);
817 let mut connector = self.connector_cache.read().unwrap().get(&key).cloned();
818 if connector.is_none() {
819 let mut cache = self.connector_cache.write().unwrap();
820 if !cache.contains_key(&key) {
822 let start = components.time_source().map(|ts| ts.now());
823 let connector = (self.connector_fn)(
824 self.client_builder.clone(),
825 Some(settings),
826 Some(components),
827 );
828 let end = components.time_source().map(|ts| ts.now());
829 if let (Some(start), Some(end)) = (start, end) {
830 if let Ok(elapsed) = end.duration_since(start) {
831 tracing::debug!("new connector created in {:?}", elapsed);
832 }
833 }
834 let connector = SharedHttpConnector::new(connector);
835 cache.insert(key.clone(), connector);
836 }
837 connector = cache.get(&key).cloned();
838 }
839
840 connector.expect("cache populated above")
841 }
842
843 fn validate_base_client_config(
844 &self,
845 _: &RuntimeComponentsBuilder,
846 _: &ConfigBag,
847 ) -> Result<(), BoxError> {
848 let _ = (self.connector_fn)(self.client_builder.clone(), None, None);
854 Ok(())
855 }
856
857 fn connector_metadata(&self) -> Option<ConnectorMetadata> {
858 Some(ConnectorMetadata::new("hyper", Some(Cow::Borrowed("1.x"))))
859 }
860}
861
862#[derive(Clone, Default, Debug)]
873pub struct Builder<Tls = TlsUnset> {
874 client_builder: Option<hyper_util::client::legacy::Builder>,
875 pool_idle_timeout: Option<Option<Duration>>,
876 pool_max_idle_per_host: Option<usize>,
877 #[allow(unused)]
878 tls_provider: Tls,
879}
880
881cfg_tls! {
882 use aws_smithy_runtime_api::client::dns::ResolveDns;
883
884 impl ConnectorBuilder<TlsProviderSelected> {
885 pub fn build(self) -> Connector {
887 let http_connector = self.base_connector();
888 self.build_https(http_connector)
889 }
890
891 pub fn tls_context(mut self, ctx: TlsContext) -> Self {
893 self.tls.context = ctx;
894 self
895 }
896
897 pub fn set_tls_context(&mut self, ctx: TlsContext) -> &mut Self {
899 self.tls.context = ctx;
900 self
901 }
902
903 pub fn build_with_resolver<R: ResolveDns + Clone + 'static>(self, resolver: R) -> Connector {
905 use crate::client::dns::HyperUtilResolver;
906 let http_connector = self.base_connector_with_resolver(HyperUtilResolver { resolver });
907 self.build_https(http_connector)
908 }
909
910 fn build_https<R>(self, http_connector: HyperHttpConnector<R>) -> Connector
911 where
912 R: Clone + Send + Sync + 'static,
913 R: tower::Service<hyper_util::client::legacy::connect::dns::Name>,
914 R::Response: Iterator<Item = std::net::SocketAddr>,
915 R::Future: Send,
916 R::Error: Into<Box<dyn Error + Send + Sync>>,
917 {
918 match &self.tls.provider {
919 #[cfg(feature = "__rustls")]
921 tls::Provider::Rustls(crypto_mode) => {
922 let proxy_config = self.proxy_config.clone()
923 .unwrap_or_else(proxy::ProxyConfig::disabled);
924
925 let https_connector = tls::rustls_provider::build_connector::wrap_connector(
926 http_connector,
927 crypto_mode.clone(),
928 &self.tls.context,
929 proxy_config,
930 );
931 self.wrap_connector(https_connector)
932 },
933 #[cfg(feature = "s2n-tls")]
934 tls::Provider::S2nTls => {
935 let proxy_config = self.proxy_config.clone()
936 .unwrap_or_else(proxy::ProxyConfig::disabled);
937
938 let https_connector = tls::s2n_tls_provider::build_connector::wrap_connector(
939 http_connector,
940 &self.tls.context,
941 proxy_config,
942 );
943 self.wrap_connector(https_connector)
944 }
945 }
946 }
947 }
948
949 impl Builder<TlsProviderSelected> {
950 pub fn build_https(self) -> SharedHttpClient {
955 build_with_conn_fn(
956 self.client_builder,
957 self.pool_idle_timeout,
958 self.pool_max_idle_per_host,
959 move |client_builder, settings, runtime_components| {
960 let builder = new_conn_builder(client_builder, settings, runtime_components)
961 .tls_provider(self.tls_provider.provider.clone())
962 .tls_context(self.tls_provider.context.clone());
963 builder.build()
964 },
965 )
966 }
967
968 pub fn build_with_resolver(
970 self,
971 resolver: impl ResolveDns + Clone + 'static,
972 ) -> SharedHttpClient {
973 build_with_conn_fn(
974 self.client_builder,
975 self.pool_idle_timeout,
976 self.pool_max_idle_per_host,
977 move |client_builder, settings, runtime_components| {
978 let builder = new_conn_builder(client_builder, settings, runtime_components)
979 .tls_provider(self.tls_provider.provider.clone())
980 .tls_context(self.tls_provider.context.clone());
981 builder.build_with_resolver(resolver.clone())
982 },
983 )
984 }
985
986 pub fn tls_context(mut self, ctx: TlsContext) -> Self {
988 self.tls_provider.context = ctx;
989 self
990 }
991 }
992}
993
994impl<Any> Builder<Any> {
995 pub fn pool_idle_timeout<D>(mut self, val: D) -> Self
1019 where
1020 D: Into<Option<Duration>>,
1021 {
1022 self.pool_idle_timeout = Some(val.into());
1023 self
1024 }
1025
1026 pub fn set_pool_idle_timeout(&mut self, val: Option<Option<Duration>>) -> &mut Self {
1049 self.pool_idle_timeout = val;
1050 self
1051 }
1052
1053 pub fn pool_max_idle_per_host(mut self, val: usize) -> Self {
1073 self.pool_max_idle_per_host = Some(val);
1074 self
1075 }
1076
1077 pub fn set_pool_max_idle_per_host(&mut self, val: Option<usize>) -> &mut Self {
1084 self.pool_max_idle_per_host = val;
1085 self
1086 }
1087}
1088
1089impl Builder<TlsUnset> {
1090 pub fn new() -> Self {
1092 Self::default()
1093 }
1094
1095 #[doc(hidden)]
1097 pub fn build_with_connector_fn<F>(self, connector_fn: F) -> SharedHttpClient
1098 where
1099 F: Fn(Option<&HttpConnectorSettings>, Option<&RuntimeComponents>) -> Connector
1100 + Send
1101 + Sync
1102 + 'static,
1103 {
1104 build_with_conn_fn(
1105 self.client_builder,
1106 self.pool_idle_timeout,
1107 self.pool_max_idle_per_host,
1108 move |_builder, settings, runtime_components| {
1109 connector_fn(settings, runtime_components)
1110 },
1111 )
1112 }
1113
1114 #[doc(hidden)]
1116 pub fn build_http(self) -> SharedHttpClient {
1117 build_with_conn_fn(
1118 self.client_builder,
1119 self.pool_idle_timeout,
1120 self.pool_max_idle_per_host,
1121 move |client_builder, settings, runtime_components| {
1122 let builder = new_conn_builder(client_builder, settings, runtime_components);
1123 builder.build_http()
1124 },
1125 )
1126 }
1127
1128 pub fn tls_provider(self, provider: tls::Provider) -> Builder<TlsProviderSelected> {
1130 Builder {
1131 client_builder: self.client_builder,
1132 pool_idle_timeout: self.pool_idle_timeout,
1133 pool_max_idle_per_host: self.pool_max_idle_per_host,
1134 tls_provider: TlsProviderSelected {
1135 provider,
1136 context: TlsContext::default(),
1137 },
1138 }
1139 }
1140}
1141
1142pub(crate) fn build_with_conn_fn<F>(
1143 client_builder: Option<hyper_util::client::legacy::Builder>,
1144 pool_idle_timeout: Option<Option<Duration>>,
1145 pool_max_idle_per_host: Option<usize>,
1146 connector_fn: F,
1147) -> SharedHttpClient
1148where
1149 F: Fn(
1150 hyper_util::client::legacy::Builder,
1151 Option<&HttpConnectorSettings>,
1152 Option<&RuntimeComponents>,
1153 ) -> Connector
1154 + Send
1155 + Sync
1156 + 'static,
1157{
1158 let client_builder = client_builder
1159 .unwrap_or_else(|| new_tokio_hyper_builder(pool_idle_timeout, pool_max_idle_per_host));
1160 SharedHttpClient::new(HyperClient {
1161 connector_cache: RwLock::new(HashMap::new()),
1162 client_builder,
1163 connector_fn,
1164 })
1165}
1166
1167#[allow(dead_code)]
1168pub(crate) fn build_with_tcp_conn_fn<C, F>(
1169 client_builder: Option<hyper_util::client::legacy::Builder>,
1170 pool_idle_timeout: Option<Option<Duration>>,
1171 pool_max_idle_per_host: Option<usize>,
1172 tcp_connector_fn: F,
1173) -> SharedHttpClient
1174where
1175 F: Fn() -> C + Send + Sync + 'static,
1176 C: Clone + Send + Sync + 'static,
1177 C: tower::Service<Uri>,
1178 C::Response: Connection + Read + Write + Send + Sync + Unpin + 'static,
1179 C::Future: Unpin + Send + 'static,
1180 C::Error: Into<BoxError>,
1181 C: Connect,
1182{
1183 build_with_conn_fn(
1184 client_builder,
1185 pool_idle_timeout,
1186 pool_max_idle_per_host,
1187 move |client_builder, settings, runtime_components| {
1188 let builder = new_conn_builder(client_builder, settings, runtime_components);
1189 builder.wrap_connector(tcp_connector_fn())
1190 },
1191 )
1192}
1193
1194fn new_conn_builder(
1195 client_builder: hyper_util::client::legacy::Builder,
1196 settings: Option<&HttpConnectorSettings>,
1197 runtime_components: Option<&RuntimeComponents>,
1198) -> ConnectorBuilder {
1199 let mut builder = Connector::builder().hyper_builder(client_builder);
1200 builder.set_connector_settings(settings.cloned());
1201 if let Some(components) = runtime_components {
1202 builder.set_sleep_impl(components.sleep_impl());
1203 if let Some(time_source) = components.time_source() {
1204 builder.set_time_source(time_source);
1205 }
1206 }
1207 builder
1208}
1209
1210#[cfg(test)]
1211mod test {
1212 use std::io::{Error, ErrorKind};
1213 use std::pin::Pin;
1214 use std::sync::atomic::{AtomicU32, Ordering};
1215 use std::sync::Arc;
1216 use std::task::{Context, Poll};
1217
1218 use crate::client::timeout::test::NeverConnects;
1219 use aws_smithy_async::assert_elapsed;
1220 use aws_smithy_async::rt::sleep::TokioSleep;
1221 use aws_smithy_async::test_util::ManualTimeSource;
1222 use aws_smithy_async::time::SystemTimeSource;
1223 use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
1224 use http_1x::Uri;
1225 use hyper::rt::ReadBufCursor;
1226 use hyper_util::client::legacy::connect::Connected;
1227 use std::time::UNIX_EPOCH;
1228
1229 #[test]
1230 fn connector_call_timer_ignores_a_backwards_clock() {
1231 let capture = CaptureHttpAttemptTelemetry::new();
1232 let time_source = SharedTimeSource::new(ManualTimeSource::new(SystemTime::UNIX_EPOCH));
1233
1234 ConnectorCallTimer {
1235 capture: capture.clone(),
1236 time_source,
1237 started_at: SystemTime::UNIX_EPOCH + Duration::from_secs(1),
1238 }
1239 .finish();
1240
1241 assert_eq!(None, capture.get().connector_call_duration());
1242 }
1243
1244 use super::*;
1245
1246 #[tokio::test]
1247 async fn connector_selection() {
1248 let creation_count = Arc::new(AtomicU32::new(0));
1250 let http_client = build_with_tcp_conn_fn(None, None, None, {
1251 let count = creation_count.clone();
1252 move || {
1253 count.fetch_add(1, Ordering::Relaxed);
1254 NeverConnects
1255 }
1256 });
1257
1258 let settings = [
1260 HttpConnectorSettings::builder()
1261 .connect_timeout(Duration::from_secs(3))
1262 .build(),
1263 HttpConnectorSettings::builder()
1264 .read_timeout(Duration::from_secs(3))
1265 .build(),
1266 HttpConnectorSettings::builder()
1267 .connect_timeout(Duration::from_secs(3))
1268 .read_timeout(Duration::from_secs(3))
1269 .build(),
1270 HttpConnectorSettings::builder()
1271 .connect_timeout(Duration::from_secs(5))
1272 .read_timeout(Duration::from_secs(3))
1273 .build(),
1274 ];
1275
1276 let components = RuntimeComponentsBuilder::for_tests()
1278 .with_time_source(Some(SystemTimeSource::new()))
1279 .build()
1280 .unwrap();
1281 let mut handles = Vec::new();
1282 for setting in &settings {
1283 for _ in 0..1000 {
1284 let client = http_client.clone();
1285 handles.push(tokio::spawn({
1286 let setting = setting.clone();
1287 let components = components.clone();
1288 async move {
1289 let _ = client.http_connector(&setting, &components);
1290 }
1291 }));
1292 }
1293 }
1294 for handle in handles {
1295 handle.await.unwrap();
1296 }
1297
1298 assert_eq!(4, creation_count.load(Ordering::Relaxed));
1300 }
1301
1302 #[tokio::test]
1303 async fn hyper_io_error() {
1304 let time = ManualTimeSource::new(UNIX_EPOCH);
1305 let connector = TestConnection {
1306 inner: HangupStream { time: time.clone() },
1307 };
1308 let mut builder = Connector::builder();
1309 builder.set_time_source(SharedTimeSource::new(time));
1310 let adapter = builder.wrap_connector(connector).adapter;
1311 let capture = CaptureHttpAttemptTelemetry::new();
1312 let mut request = HttpRequest::get("https://socket-hangup.com").unwrap();
1313 request.add_extension(capture.clone());
1314 let err = adapter.call(request).await.expect_err("socket hangup");
1315 assert!(err.is_io(), "unexpected error type: {:?}", err);
1316 assert_eq!(
1317 capture.get().connector_call_duration(),
1318 Some(Duration::from_secs(7))
1319 );
1320 }
1321
1322 #[derive(Clone)]
1324 struct HangupStream {
1325 time: ManualTimeSource,
1326 }
1327
1328 impl Connection for HangupStream {
1329 fn connected(&self) -> Connected {
1330 Connected::new()
1331 }
1332 }
1333
1334 impl Read for HangupStream {
1335 fn poll_read(
1336 self: Pin<&mut Self>,
1337 _cx: &mut Context<'_>,
1338 _buf: ReadBufCursor<'_>,
1339 ) -> Poll<std::io::Result<()>> {
1340 self.get_mut().time.advance(Duration::from_secs(7));
1341 Poll::Ready(Err(Error::new(
1342 ErrorKind::ConnectionReset,
1343 "connection reset",
1344 )))
1345 }
1346 }
1347
1348 impl Write for HangupStream {
1349 fn poll_write(
1350 self: Pin<&mut Self>,
1351 _cx: &mut Context<'_>,
1352 _buf: &[u8],
1353 ) -> Poll<Result<usize, Error>> {
1354 Poll::Pending
1355 }
1356
1357 fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1358 Poll::Pending
1359 }
1360
1361 fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1362 Poll::Pending
1363 }
1364 }
1365
1366 #[derive(Clone)]
1367 struct TestConnection<T> {
1368 inner: T,
1369 }
1370
1371 impl<T> tower::Service<Uri> for TestConnection<T>
1372 where
1373 T: Clone + Connection,
1374 {
1375 type Response = T;
1376 type Error = BoxError;
1377 type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
1378
1379 fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
1380 Poll::Ready(Ok(()))
1381 }
1382
1383 fn call(&mut self, _req: Uri) -> Self::Future {
1384 std::future::ready(Ok(self.inner.clone()))
1385 }
1386 }
1387
1388 #[tokio::test]
1389 async fn http_connect_timeout_works() {
1390 let tcp_connector = NeverConnects::default();
1391 let connector_settings = HttpConnectorSettings::builder()
1392 .connect_timeout(Duration::from_secs(1))
1393 .build();
1394 let hyper = Connector::builder()
1395 .connector_settings(connector_settings)
1396 .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1397 .wrap_connector(tcp_connector)
1398 .adapter;
1399 let now = tokio::time::Instant::now();
1400 tokio::time::pause();
1401 let resp = hyper
1402 .call(HttpRequest::get("https://static-uri.com").unwrap())
1403 .await
1404 .unwrap_err();
1405 assert!(
1406 resp.is_timeout(),
1407 "expected resp.is_timeout() to be true but it was false, resp == {:?}",
1408 resp
1409 );
1410 let message = DisplayErrorContext(&resp).to_string();
1411 let expected = "timeout: client error (Connect): HTTP connect timeout occurred after 1s";
1412 assert!(
1413 message.contains(expected),
1414 "expected '{message}' to contain '{expected}'"
1415 );
1416 assert_elapsed!(now, Duration::from_secs(1));
1417 }
1418
1419 #[tokio::test]
1420 async fn http_read_timeout_works() {
1421 let tcp_connector = crate::client::timeout::test::NeverReplies;
1422 let connector_settings = HttpConnectorSettings::builder()
1423 .connect_timeout(Duration::from_secs(1))
1424 .read_timeout(Duration::from_secs(2))
1425 .build();
1426 let hyper = Connector::builder()
1427 .connector_settings(connector_settings)
1428 .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1429 .wrap_connector(tcp_connector)
1430 .adapter;
1431 let now = tokio::time::Instant::now();
1432 tokio::time::pause();
1433 let err = hyper
1434 .call(HttpRequest::get("https://fake-uri.com").unwrap())
1435 .await
1436 .unwrap_err();
1437 assert!(
1438 err.is_timeout(),
1439 "expected err.is_timeout() to be true but it was false, err == {err:?}",
1440 );
1441 let message = format!("{}", DisplayErrorContext(&err));
1442 let expected = "timeout: HTTP read timeout occurred after 2s";
1443 assert!(
1444 message.contains(expected),
1445 "expected '{message}' to contain '{expected}'"
1446 );
1447 assert_elapsed!(now, Duration::from_secs(2));
1448 }
1449
1450 #[cfg(not(windows))]
1451 #[tokio::test]
1452 async fn connection_refused_works() {
1453 use crate::client::dns::HyperUtilResolver;
1454 use aws_smithy_runtime_api::client::dns::{DnsFuture, ResolveDns};
1455 use std::net::{IpAddr, Ipv4Addr};
1456
1457 #[derive(Debug, Clone, Default)]
1458 struct TestResolver;
1459 impl ResolveDns for TestResolver {
1460 fn resolve_dns<'a>(&'a self, _name: &'a str) -> DnsFuture<'a> {
1461 let localhost_v4 = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
1462 DnsFuture::ready(Ok(vec![localhost_v4]))
1463 }
1464 }
1465
1466 let connector_settings = HttpConnectorSettings::builder()
1467 .connect_timeout(Duration::from_secs(20))
1468 .build();
1469
1470 let resolver = HyperUtilResolver {
1471 resolver: TestResolver,
1472 };
1473 let connector = Connector::builder().base_connector_with_resolver(resolver);
1474
1475 let hyper = Connector::builder()
1476 .connector_settings(connector_settings)
1477 .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1478 .wrap_connector(connector)
1479 .adapter;
1480
1481 let resp = hyper
1482 .call(HttpRequest::get("http://static-uri:50227.com").unwrap())
1483 .await
1484 .unwrap_err();
1485 assert!(
1486 resp.is_io(),
1487 "expected resp.is_io() to be true but it was false, resp == {:?}",
1488 resp
1489 );
1490 let message = DisplayErrorContext(&resp).to_string();
1491 let expected = "Connection refused";
1492 assert!(
1493 message.contains(expected),
1494 "expected '{message}' to contain '{expected}'"
1495 );
1496 }
1497
1498 #[cfg(feature = "s2n-tls")]
1499 #[tokio::test]
1500 async fn s2n_tls_provider() {
1501 let client = Builder::new()
1503 .tls_provider(tls::Provider::S2nTls)
1504 .build_https();
1505 let connector_settings = HttpConnectorSettings::builder().build();
1506
1507 let runtime_components = RuntimeComponentsBuilder::for_tests()
1510 .with_time_source(Some(SystemTimeSource::new()))
1511 .build()
1512 .unwrap();
1513
1514 let connector = client.http_connector(&connector_settings, &runtime_components);
1515
1516 let error = connector
1521 .call(HttpRequest::get("notascheme://amazon.com").unwrap())
1522 .await
1523 .unwrap_err();
1524 let error = error.into_source();
1525 let s2n_error = find_source::<s2n_tls_hyper::error::Error>(error.as_ref())
1526 .expect("s2n-tls-hyper error is present in the source chain");
1527 assert!(matches!(
1528 s2n_error,
1529 s2n_tls_hyper::error::Error::InvalidScheme
1530 ));
1531 }
1532
1533 #[test]
1535 fn connector_metadata_identifies_hyper_1x() {
1536 let client = crate::Builder::new().build_http();
1537 let metadata = client
1538 .connector_metadata()
1539 .expect("connector metadata should be present");
1540 assert_eq!(metadata.name(), Cow::Borrowed("hyper"));
1541 assert_eq!(metadata.version(), Some(Cow::Borrowed("1.x")));
1542 }
1543}