1mod dns;
7pub mod proxy;
9mod timeout;
10pub mod tls;
12
13pub(crate) mod connect;
14
15use crate::cfg::cfg_tls;
16use crate::tls::TlsContext;
17use aws_smithy_async::future::timeout::TimedOutError;
18use aws_smithy_async::rt::sleep::{default_async_sleep, AsyncSleep, SharedAsyncSleep};
19use aws_smithy_runtime_api::box_error::BoxError;
20use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
21use aws_smithy_runtime_api::client::connection::ConnectionMetadata;
22use aws_smithy_runtime_api::client::connector_metadata::ConnectorMetadata;
23use aws_smithy_runtime_api::client::http::{
24 HttpClient, HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpClient,
25 SharedHttpConnector,
26};
27use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};
28use aws_smithy_runtime_api::client::result::ConnectorError;
29use aws_smithy_runtime_api::client::runtime_components::{
30 RuntimeComponents, RuntimeComponentsBuilder,
31};
32use aws_smithy_runtime_api::shared::IntoShared;
33use aws_smithy_types::body::SdkBody;
34use aws_smithy_types::config_bag::ConfigBag;
35use aws_smithy_types::error::display::DisplayErrorContext;
36use aws_smithy_types::retry::ErrorKind;
37use client::connect::Connection;
38use h2::Reason;
39use http_1x::{Extensions, Uri};
40use hyper::rt::{Read, Write};
41use hyper_util::client::legacy as client;
42use hyper_util::client::legacy::connect::dns::GaiResolver;
43use hyper_util::client::legacy::connect::{
44 capture_connection, CaptureConnection, Connect, HttpConnector as HyperHttpConnector, HttpInfo,
45};
46use hyper_util::client::proxy::matcher::Matcher;
47use hyper_util::rt::{TokioExecutor, TokioTimer};
48use std::borrow::Cow;
49use std::collections::HashMap;
50use std::error::Error;
51use std::fmt;
52use std::sync::RwLock;
53use std::time::Duration;
54
55pub fn default_connector(
57 settings: &HttpConnectorSettings,
58 sleep: Option<SharedAsyncSleep>,
59) -> Option<SharedHttpConnector> {
60 #[cfg(feature = "rustls-aws-lc")]
61 {
62 tracing::trace!(settings = ?settings, sleep = ?sleep, "creating a new default connector");
63 let mut conn_builder = Connector::builder().connector_settings(settings.clone());
64
65 if let Some(sleep) = sleep {
66 conn_builder = conn_builder.sleep_impl(sleep);
67 }
68
69 let conn = conn_builder
70 .tls_provider(tls::Provider::Rustls(
71 tls::rustls_provider::CryptoMode::AwsLc,
72 ))
73 .build();
74 Some(SharedHttpConnector::new(conn))
75 }
76 #[cfg(not(feature = "rustls-aws-lc"))]
77 {
78 tracing::trace!(settings = ?settings, sleep = ?sleep, "no default connector available");
79 None
80 }
81}
82
83#[derive(Debug)]
90pub struct Connector {
91 adapter: Box<dyn HttpConnector>,
92}
93
94impl Connector {
95 pub fn builder() -> ConnectorBuilder {
97 ConnectorBuilder::default()
98 }
99}
100
101impl HttpConnector for Connector {
102 fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
103 self.adapter.call(request)
104 }
105}
106
107#[derive(Debug, Clone)]
109pub struct ConnectorBuilder<Tls = TlsUnset> {
110 connector_settings: Option<HttpConnectorSettings>,
111 sleep_impl: Option<SharedAsyncSleep>,
112 client_builder: Option<hyper_util::client::legacy::Builder>,
113 pool_idle_timeout: Option<Option<Duration>>,
114 pool_max_idle_per_host: Option<usize>,
115 enable_tcp_nodelay: bool,
116 interface: Option<String>,
117 proxy_config: Option<proxy::ProxyConfig>,
118 #[allow(unused)]
119 tls: Tls,
120}
121
122impl<Tls: Default> Default for ConnectorBuilder<Tls> {
123 fn default() -> Self {
124 Self {
125 connector_settings: None,
126 sleep_impl: None,
127 client_builder: None,
128 pool_idle_timeout: None,
129 pool_max_idle_per_host: None,
130 enable_tcp_nodelay: true,
136 interface: None,
137 proxy_config: None,
138 tls: Tls::default(),
139 }
140 }
141}
142
143#[derive(Default, Debug, Clone)]
145#[non_exhaustive]
146pub struct TlsUnset {}
147
148#[derive(Debug, Clone)]
150pub struct TlsProviderSelected {
151 #[allow(unused)]
152 provider: tls::Provider,
153 #[allow(unused)]
154 context: TlsContext,
155}
156
157impl ConnectorBuilder<TlsUnset> {
158 pub fn tls_provider(self, provider: tls::Provider) -> ConnectorBuilder<TlsProviderSelected> {
160 ConnectorBuilder {
161 connector_settings: self.connector_settings,
162 sleep_impl: self.sleep_impl,
163 client_builder: self.client_builder,
164 enable_tcp_nodelay: self.enable_tcp_nodelay,
165 interface: self.interface,
166 proxy_config: self.proxy_config,
167 pool_idle_timeout: self.pool_idle_timeout,
168 pool_max_idle_per_host: self.pool_max_idle_per_host,
169 tls: TlsProviderSelected {
170 provider,
171 context: TlsContext::default(),
172 },
173 }
174 }
175
176 #[doc(hidden)]
178 pub fn build_http(self) -> Connector {
179 if let Some(ref proxy_config) = self.proxy_config {
180 if proxy_config.requires_tls() {
181 tracing::warn!(
182 "HTTPS proxy configured but no TLS provider set. \
183 Connections to HTTPS proxy servers will fail. \
184 Consider configuring a TLS provider to enable TLS support."
185 );
186 }
187 }
188
189 let base = self.base_connector();
190
191 let proxy_config = self
193 .proxy_config
194 .clone()
195 .unwrap_or_else(proxy::ProxyConfig::disabled);
196
197 if !proxy_config.is_disabled() {
198 let http_proxy_connector = connect::HttpProxyConnector::new(base, proxy_config);
199 self.wrap_connector(http_proxy_connector)
200 } else {
201 self.wrap_connector(base)
202 }
203 }
204}
205
206impl<Any> ConnectorBuilder<Any> {
207 pub(crate) fn wrap_connector<C>(self, tcp_connector: C) -> Connector
209 where
210 C: Send + Sync + 'static,
211 C: Clone,
212 C: tower::Service<Uri>,
213 C::Response: Read + Write + Connection + Send + Sync + Unpin,
214 C: Connect,
215 C::Future: Unpin + Send + 'static,
216 C::Error: Into<BoxError>,
217 {
218 let client_builder = self.client_builder.unwrap_or_else(|| {
219 new_tokio_hyper_builder(self.pool_idle_timeout, self.pool_max_idle_per_host)
220 });
221 let sleep_impl = self.sleep_impl.or_else(default_async_sleep);
222 let (connect_timeout, read_timeout) = self
223 .connector_settings
224 .map(|c| (c.connect_timeout(), c.read_timeout()))
225 .unwrap_or((None, None));
226
227 let connector = match connect_timeout {
228 Some(duration) => timeout::ConnectTimeout::new(
229 tcp_connector,
230 sleep_impl
231 .clone()
232 .expect("a sleep impl must be provided in order to have a connect timeout"),
233 duration,
234 ),
235 None => timeout::ConnectTimeout::no_timeout(tcp_connector),
236 };
237 let base = client_builder.build(connector);
238 let read_timeout = match read_timeout {
239 Some(duration) => timeout::HttpReadTimeout::new(
240 base,
241 sleep_impl.expect("a sleep impl must be provided in order to have a read timeout"),
242 duration,
243 ),
244 None => timeout::HttpReadTimeout::no_timeout(base),
245 };
246
247 let proxy_matcher = self
248 .proxy_config
249 .as_ref()
250 .map(|config| config.clone().into_hyper_util_matcher());
251
252 Connector {
253 adapter: Box::new(Adapter {
254 client: read_timeout,
255 proxy_matcher,
256 }),
257 }
258 }
259
260 fn base_connector(&self) -> HyperHttpConnector {
263 self.base_connector_with_resolver(GaiResolver::new())
264 }
265
266 fn base_connector_with_resolver<R>(&self, resolver: R) -> HyperHttpConnector<R> {
269 let mut conn = HyperHttpConnector::new_with_resolver(resolver);
270 conn.set_nodelay(self.enable_tcp_nodelay);
271 #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
272 if let Some(interface) = &self.interface {
273 conn.set_interface(interface);
274 }
275 conn
276 }
277
278 pub fn sleep_impl(mut self, sleep_impl: impl AsyncSleep + 'static) -> Self {
283 self.sleep_impl = Some(sleep_impl.into_shared());
284 self
285 }
286
287 pub fn set_sleep_impl(&mut self, sleep_impl: Option<SharedAsyncSleep>) -> &mut Self {
292 self.sleep_impl = sleep_impl;
293 self
294 }
295
296 pub fn connector_settings(mut self, connector_settings: HttpConnectorSettings) -> Self {
298 self.connector_settings = Some(connector_settings);
299 self
300 }
301
302 pub fn set_connector_settings(
304 &mut self,
305 connector_settings: Option<HttpConnectorSettings>,
306 ) -> &mut Self {
307 self.connector_settings = connector_settings;
308 self
309 }
310
311 pub fn enable_tcp_nodelay(mut self, nodelay: bool) -> Self {
313 self.enable_tcp_nodelay = nodelay;
314 self
315 }
316
317 pub fn set_enable_tcp_nodelay(&mut self, nodelay: bool) -> &mut Self {
319 self.enable_tcp_nodelay = nodelay;
320 self
321 }
322
323 #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
336 pub fn set_interface<S: Into<String>>(&mut self, interface: S) -> &mut Self {
337 self.interface = Some(interface.into());
338 self
339 }
340
341 pub fn proxy_config(mut self, config: proxy::ProxyConfig) -> Self {
363 self.proxy_config = Some(config);
364 self
365 }
366
367 pub fn set_proxy_config(&mut self, config: Option<proxy::ProxyConfig>) -> &mut Self {
371 self.proxy_config = config;
372 self
373 }
374
375 pub fn pool_idle_timeout<D>(mut self, val: D) -> Self
399 where
400 D: Into<Option<Duration>>,
401 {
402 self.pool_idle_timeout = Some(val.into());
403 self
404 }
405
406 pub fn set_pool_idle_timeout(&mut self, val: Option<Option<Duration>>) -> &mut Self {
430 self.pool_idle_timeout = val;
431 self
432 }
433
434 pub fn pool_max_idle_per_host(mut self, val: usize) -> Self {
454 self.pool_max_idle_per_host = Some(val);
455 self
456 }
457
458 pub fn set_pool_max_idle_per_host(&mut self, val: Option<usize>) -> &mut Self {
465 self.pool_max_idle_per_host = val;
466 self
467 }
468
469 pub(crate) fn hyper_builder(
473 mut self,
474 hyper_builder: hyper_util::client::legacy::Builder,
475 ) -> Self {
476 self.set_hyper_builder(Some(hyper_builder));
477 self
478 }
479
480 pub(crate) fn set_hyper_builder(
484 &mut self,
485 hyper_builder: Option<hyper_util::client::legacy::Builder>,
486 ) -> &mut Self {
487 self.client_builder = hyper_builder;
488 self
489 }
490}
491
492struct Adapter<C> {
496 client: timeout::HttpReadTimeout<
497 hyper_util::client::legacy::Client<timeout::ConnectTimeout<C>, SdkBody>,
498 >,
499 proxy_matcher: Option<Matcher>,
500}
501
502impl<C> fmt::Debug for Adapter<C> {
503 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
504 f.debug_struct("Adapter")
505 .field("client", &"** hyper client **")
506 .field("proxy_matcher", &self.proxy_matcher.is_some())
507 .finish()
508 }
509}
510
511fn extract_smithy_connection(capture_conn: &CaptureConnection) -> Option<ConnectionMetadata> {
513 let capture_conn = capture_conn.clone();
514 if let Some(conn) = capture_conn.clone().connection_metadata().as_ref() {
515 let mut extensions = Extensions::new();
516 conn.get_extras(&mut extensions);
517 let http_info = extensions.get::<HttpInfo>();
518 let mut builder = ConnectionMetadata::builder()
519 .proxied(conn.is_proxied())
520 .poison_fn(move || match capture_conn.connection_metadata().as_ref() {
521 Some(conn) => conn.poison(),
522 None => tracing::trace!("no connection existed to poison"),
523 });
524
525 builder
526 .set_local_addr(http_info.map(|info| info.local_addr()))
527 .set_remote_addr(http_info.map(|info| info.remote_addr()));
528
529 let smithy_connection = builder.build();
530
531 Some(smithy_connection)
532 } else {
533 None
534 }
535}
536
537fn new_tokio_hyper_builder(
538 pool_idle_timeout: Option<Option<Duration>>,
539 pool_max_idle_per_host: Option<usize>,
540) -> hyper_util::client::legacy::Builder {
541 let mut builder = hyper_util::client::legacy::Builder::new(TokioExecutor::new());
542 builder.pool_timer(TokioTimer::new());
544
545 if let Some(pool_idle_timeout) = pool_idle_timeout {
546 builder.pool_idle_timeout(pool_idle_timeout);
547 }
548
549 if let Some(max_idle) = pool_max_idle_per_host {
550 builder.pool_max_idle_per_host(max_idle);
551 }
552
553 builder
554}
555
556impl<C> Adapter<C> {
557 fn add_proxy_auth_header(&self, request: &mut http_1x::Request<SdkBody>) {
559 if request.uri().scheme() != Some(&http_1x::uri::Scheme::HTTP) {
561 return;
562 }
563
564 if request
566 .headers()
567 .contains_key(http_1x::header::PROXY_AUTHORIZATION)
568 {
569 return;
570 }
571
572 if let Some(ref matcher) = self.proxy_matcher {
573 if let Some(intercept) = matcher.intercept(request.uri()) {
574 if let Some(auth_header) = intercept.basic_auth() {
576 request
577 .headers_mut()
578 .insert(http_1x::header::PROXY_AUTHORIZATION, auth_header.clone());
579 tracing::debug!("added proxy authentication header for {}", request.uri());
580 }
581 }
582 }
583 }
584}
585
586impl<C> HttpConnector for Adapter<C>
587where
588 C: Clone + Send + Sync + 'static,
589 C: tower::Service<Uri>,
590 C::Response: Connection + Read + Write + Unpin + 'static,
591 timeout::ConnectTimeout<C>: Connect,
592 C::Future: Unpin + Send + 'static,
593 C::Error: Into<BoxError>,
594{
595 fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
596 let mut request = match request.try_into_http1x() {
597 Ok(request) => request,
598 Err(err) => {
599 return HttpConnectorFuture::ready(Err(ConnectorError::user(err.into())));
600 }
601 };
602
603 self.add_proxy_auth_header(&mut request);
604
605 let capture_connection = capture_connection(&mut request);
606 if let Some(capture_smithy_connection) =
607 request.extensions().get::<CaptureSmithyConnection>()
608 {
609 capture_smithy_connection
610 .set_connection_retriever(move || extract_smithy_connection(&capture_connection));
611 }
612 let mut client = self.client.clone();
613 use tower::Service;
614 let fut = client.call(request);
615 HttpConnectorFuture::new(async move {
616 let response = fut
617 .await
618 .map_err(downcast_error)?
619 .map(SdkBody::from_body_1_x);
620 match HttpResponse::try_from(response) {
621 Ok(response) => Ok(response),
622 Err(err) => Err(ConnectorError::other(err.into(), None)),
623 }
624 })
625 }
626}
627
628fn downcast_error(err: BoxError) -> ConnectorError {
630 if find_source::<TimedOutError>(err.as_ref()).is_some() {
632 return ConnectorError::timeout(err);
633 }
634 let err = match err.downcast::<ConnectorError>() {
636 Ok(connector_error) => return *connector_error,
637 Err(box_error) => box_error,
638 };
639 let err = match find_source::<hyper::Error>(err.as_ref()) {
642 Some(hyper_error) => return to_connector_error(hyper_error)(err),
643 None => match find_source::<hyper_util::client::legacy::Error>(err.as_ref()) {
644 Some(hyper_util_err) => {
645 if hyper_util_err.is_connect()
646 || find_source::<std::io::Error>(hyper_util_err).is_some()
647 {
648 return ConnectorError::io(err);
649 }
650 err
651 }
652 None => err,
653 },
654 };
655
656 ConnectorError::other(err, None)
658}
659
660fn to_connector_error(err: &hyper::Error) -> fn(BoxError) -> ConnectorError {
662 if err.is_timeout() || find_source::<timeout::HttpTimeoutError>(err).is_some() {
663 return ConnectorError::timeout;
664 }
665 if err.is_user() {
666 return ConnectorError::user;
667 }
668 if err.is_closed() || err.is_canceled() || find_source::<std::io::Error>(err).is_some() {
669 return ConnectorError::io;
670 }
671 if err.is_incomplete_message() {
673 return |err: BoxError| ConnectorError::other(err, Some(ErrorKind::TransientError));
674 }
675
676 if let Some(h2_err) = find_source::<h2::Error>(err) {
677 if h2_err.is_go_away()
678 || (h2_err.is_reset() && h2_err.reason() == Some(Reason::REFUSED_STREAM))
679 {
680 return ConnectorError::io;
681 }
682 }
683
684 tracing::warn!(err = %DisplayErrorContext(&err), "unrecognized error from Hyper. If this error should be retried, please file an issue.");
685 |err: BoxError| ConnectorError::other(err, None)
686}
687
688fn find_source<'a, E: Error + 'static>(err: &'a (dyn Error + 'static)) -> Option<&'a E> {
689 let mut next = Some(err);
690 while let Some(err) = next {
691 if let Some(matching_err) = err.downcast_ref::<E>() {
692 return Some(matching_err);
693 }
694 next = err.source();
695 }
696 None
697}
698
699#[derive(Clone, Debug, Eq, PartialEq, Hash)]
703struct CacheKey {
704 connect_timeout: Option<Duration>,
705 read_timeout: Option<Duration>,
706}
707
708impl From<&HttpConnectorSettings> for CacheKey {
709 fn from(value: &HttpConnectorSettings) -> Self {
710 Self {
711 connect_timeout: value.connect_timeout(),
712 read_timeout: value.read_timeout(),
713 }
714 }
715}
716
717struct HyperClient<F> {
718 connector_cache: RwLock<HashMap<CacheKey, SharedHttpConnector>>,
719 client_builder: hyper_util::client::legacy::Builder,
720 connector_fn: F,
721}
722
723impl<F> fmt::Debug for HyperClient<F> {
724 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
725 f.debug_struct("HyperClient")
726 .field("connector_cache", &self.connector_cache)
727 .field("client_builder", &self.client_builder)
728 .finish()
729 }
730}
731
732impl<F> HttpClient for HyperClient<F>
733where
734 F: Fn(
735 hyper_util::client::legacy::Builder,
736 Option<&HttpConnectorSettings>,
737 Option<&RuntimeComponents>,
738 ) -> Connector
739 + Send
740 + Sync
741 + 'static,
742{
743 fn http_connector(
744 &self,
745 settings: &HttpConnectorSettings,
746 components: &RuntimeComponents,
747 ) -> SharedHttpConnector {
748 let key = CacheKey::from(settings);
749 let mut connector = self.connector_cache.read().unwrap().get(&key).cloned();
750 if connector.is_none() {
751 let mut cache = self.connector_cache.write().unwrap();
752 if !cache.contains_key(&key) {
754 let start = components.time_source().map(|ts| ts.now());
755 let connector = (self.connector_fn)(
756 self.client_builder.clone(),
757 Some(settings),
758 Some(components),
759 );
760 let end = components.time_source().map(|ts| ts.now());
761 if let (Some(start), Some(end)) = (start, end) {
762 if let Ok(elapsed) = end.duration_since(start) {
763 tracing::debug!("new connector created in {:?}", elapsed);
764 }
765 }
766 let connector = SharedHttpConnector::new(connector);
767 cache.insert(key.clone(), connector);
768 }
769 connector = cache.get(&key).cloned();
770 }
771
772 connector.expect("cache populated above")
773 }
774
775 fn validate_base_client_config(
776 &self,
777 _: &RuntimeComponentsBuilder,
778 _: &ConfigBag,
779 ) -> Result<(), BoxError> {
780 let _ = (self.connector_fn)(self.client_builder.clone(), None, None);
786 Ok(())
787 }
788
789 fn connector_metadata(&self) -> Option<ConnectorMetadata> {
790 Some(ConnectorMetadata::new("hyper", Some(Cow::Borrowed("1.x"))))
791 }
792}
793
794#[derive(Clone, Default, Debug)]
805pub struct Builder<Tls = TlsUnset> {
806 client_builder: Option<hyper_util::client::legacy::Builder>,
807 pool_idle_timeout: Option<Option<Duration>>,
808 pool_max_idle_per_host: Option<usize>,
809 #[allow(unused)]
810 tls_provider: Tls,
811}
812
813cfg_tls! {
814 use aws_smithy_runtime_api::client::dns::ResolveDns;
815
816 impl ConnectorBuilder<TlsProviderSelected> {
817 pub fn build(self) -> Connector {
819 let http_connector = self.base_connector();
820 self.build_https(http_connector)
821 }
822
823 pub fn tls_context(mut self, ctx: TlsContext) -> Self {
825 self.tls.context = ctx;
826 self
827 }
828
829 pub fn set_tls_context(&mut self, ctx: TlsContext) -> &mut Self {
831 self.tls.context = ctx;
832 self
833 }
834
835 pub fn build_with_resolver<R: ResolveDns + Clone + 'static>(self, resolver: R) -> Connector {
837 use crate::client::dns::HyperUtilResolver;
838 let http_connector = self.base_connector_with_resolver(HyperUtilResolver { resolver });
839 self.build_https(http_connector)
840 }
841
842 fn build_https<R>(self, http_connector: HyperHttpConnector<R>) -> Connector
843 where
844 R: Clone + Send + Sync + 'static,
845 R: tower::Service<hyper_util::client::legacy::connect::dns::Name>,
846 R::Response: Iterator<Item = std::net::SocketAddr>,
847 R::Future: Send,
848 R::Error: Into<Box<dyn Error + Send + Sync>>,
849 {
850 match &self.tls.provider {
851 #[cfg(feature = "__rustls")]
853 tls::Provider::Rustls(crypto_mode) => {
854 let proxy_config = self.proxy_config.clone()
855 .unwrap_or_else(proxy::ProxyConfig::disabled);
856
857 let https_connector = tls::rustls_provider::build_connector::wrap_connector(
858 http_connector,
859 crypto_mode.clone(),
860 &self.tls.context,
861 proxy_config,
862 );
863 self.wrap_connector(https_connector)
864 },
865 #[cfg(feature = "s2n-tls")]
866 tls::Provider::S2nTls => {
867 let proxy_config = self.proxy_config.clone()
868 .unwrap_or_else(proxy::ProxyConfig::disabled);
869
870 let https_connector = tls::s2n_tls_provider::build_connector::wrap_connector(
871 http_connector,
872 &self.tls.context,
873 proxy_config,
874 );
875 self.wrap_connector(https_connector)
876 }
877 }
878 }
879 }
880
881 impl Builder<TlsProviderSelected> {
882 pub fn build_https(self) -> SharedHttpClient {
887 build_with_conn_fn(
888 self.client_builder,
889 self.pool_idle_timeout,
890 self.pool_max_idle_per_host,
891 move |client_builder, settings, runtime_components| {
892 let builder = new_conn_builder(client_builder, settings, runtime_components)
893 .tls_provider(self.tls_provider.provider.clone())
894 .tls_context(self.tls_provider.context.clone());
895 builder.build()
896 },
897 )
898 }
899
900 pub fn build_with_resolver(
902 self,
903 resolver: impl ResolveDns + Clone + 'static,
904 ) -> SharedHttpClient {
905 build_with_conn_fn(
906 self.client_builder,
907 self.pool_idle_timeout,
908 self.pool_max_idle_per_host,
909 move |client_builder, settings, runtime_components| {
910 let builder = new_conn_builder(client_builder, settings, runtime_components)
911 .tls_provider(self.tls_provider.provider.clone())
912 .tls_context(self.tls_provider.context.clone());
913 builder.build_with_resolver(resolver.clone())
914 },
915 )
916 }
917
918 pub fn tls_context(mut self, ctx: TlsContext) -> Self {
920 self.tls_provider.context = ctx;
921 self
922 }
923 }
924}
925
926impl<Any> Builder<Any> {
927 pub fn pool_idle_timeout<D>(mut self, val: D) -> Self
951 where
952 D: Into<Option<Duration>>,
953 {
954 self.pool_idle_timeout = Some(val.into());
955 self
956 }
957
958 pub fn set_pool_idle_timeout(&mut self, val: Option<Option<Duration>>) -> &mut Self {
981 self.pool_idle_timeout = val;
982 self
983 }
984
985 pub fn pool_max_idle_per_host(mut self, val: usize) -> Self {
1005 self.pool_max_idle_per_host = Some(val);
1006 self
1007 }
1008
1009 pub fn set_pool_max_idle_per_host(&mut self, val: Option<usize>) -> &mut Self {
1016 self.pool_max_idle_per_host = val;
1017 self
1018 }
1019}
1020
1021impl Builder<TlsUnset> {
1022 pub fn new() -> Self {
1024 Self::default()
1025 }
1026
1027 #[doc(hidden)]
1029 pub fn build_with_connector_fn<F>(self, connector_fn: F) -> SharedHttpClient
1030 where
1031 F: Fn(Option<&HttpConnectorSettings>, Option<&RuntimeComponents>) -> Connector
1032 + Send
1033 + Sync
1034 + 'static,
1035 {
1036 build_with_conn_fn(
1037 self.client_builder,
1038 self.pool_idle_timeout,
1039 self.pool_max_idle_per_host,
1040 move |_builder, settings, runtime_components| {
1041 connector_fn(settings, runtime_components)
1042 },
1043 )
1044 }
1045
1046 #[doc(hidden)]
1048 pub fn build_http(self) -> SharedHttpClient {
1049 build_with_conn_fn(
1050 self.client_builder,
1051 self.pool_idle_timeout,
1052 self.pool_max_idle_per_host,
1053 move |client_builder, settings, runtime_components| {
1054 let builder = new_conn_builder(client_builder, settings, runtime_components);
1055 builder.build_http()
1056 },
1057 )
1058 }
1059
1060 pub fn tls_provider(self, provider: tls::Provider) -> Builder<TlsProviderSelected> {
1062 Builder {
1063 client_builder: self.client_builder,
1064 pool_idle_timeout: self.pool_idle_timeout,
1065 pool_max_idle_per_host: self.pool_max_idle_per_host,
1066 tls_provider: TlsProviderSelected {
1067 provider,
1068 context: TlsContext::default(),
1069 },
1070 }
1071 }
1072}
1073
1074pub(crate) fn build_with_conn_fn<F>(
1075 client_builder: Option<hyper_util::client::legacy::Builder>,
1076 pool_idle_timeout: Option<Option<Duration>>,
1077 pool_max_idle_per_host: Option<usize>,
1078 connector_fn: F,
1079) -> SharedHttpClient
1080where
1081 F: Fn(
1082 hyper_util::client::legacy::Builder,
1083 Option<&HttpConnectorSettings>,
1084 Option<&RuntimeComponents>,
1085 ) -> Connector
1086 + Send
1087 + Sync
1088 + 'static,
1089{
1090 let client_builder = client_builder
1091 .unwrap_or_else(|| new_tokio_hyper_builder(pool_idle_timeout, pool_max_idle_per_host));
1092 SharedHttpClient::new(HyperClient {
1093 connector_cache: RwLock::new(HashMap::new()),
1094 client_builder,
1095 connector_fn,
1096 })
1097}
1098
1099#[allow(dead_code)]
1100pub(crate) fn build_with_tcp_conn_fn<C, F>(
1101 client_builder: Option<hyper_util::client::legacy::Builder>,
1102 pool_idle_timeout: Option<Option<Duration>>,
1103 pool_max_idle_per_host: Option<usize>,
1104 tcp_connector_fn: F,
1105) -> SharedHttpClient
1106where
1107 F: Fn() -> C + Send + Sync + 'static,
1108 C: Clone + Send + Sync + 'static,
1109 C: tower::Service<Uri>,
1110 C::Response: Connection + Read + Write + Send + Sync + Unpin + 'static,
1111 C::Future: Unpin + Send + 'static,
1112 C::Error: Into<BoxError>,
1113 C: Connect,
1114{
1115 build_with_conn_fn(
1116 client_builder,
1117 pool_idle_timeout,
1118 pool_max_idle_per_host,
1119 move |client_builder, settings, runtime_components| {
1120 let builder = new_conn_builder(client_builder, settings, runtime_components);
1121 builder.wrap_connector(tcp_connector_fn())
1122 },
1123 )
1124}
1125
1126fn new_conn_builder(
1127 client_builder: hyper_util::client::legacy::Builder,
1128 settings: Option<&HttpConnectorSettings>,
1129 runtime_components: Option<&RuntimeComponents>,
1130) -> ConnectorBuilder {
1131 let mut builder = Connector::builder().hyper_builder(client_builder);
1132 builder.set_connector_settings(settings.cloned());
1133 if let Some(components) = runtime_components {
1134 builder.set_sleep_impl(components.sleep_impl());
1135 }
1136 builder
1137}
1138
1139#[cfg(test)]
1140mod test {
1141 use std::io::{Error, ErrorKind};
1142 use std::pin::Pin;
1143 use std::sync::atomic::{AtomicU32, Ordering};
1144 use std::sync::Arc;
1145 use std::task::{Context, Poll};
1146
1147 use crate::client::timeout::test::NeverConnects;
1148 use aws_smithy_async::assert_elapsed;
1149 use aws_smithy_async::rt::sleep::TokioSleep;
1150 use aws_smithy_async::time::SystemTimeSource;
1151 use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
1152 use http_1x::Uri;
1153 use hyper::rt::ReadBufCursor;
1154 use hyper_util::client::legacy::connect::Connected;
1155
1156 use super::*;
1157
1158 #[tokio::test]
1159 async fn connector_selection() {
1160 let creation_count = Arc::new(AtomicU32::new(0));
1162 let http_client = build_with_tcp_conn_fn(None, None, None, {
1163 let count = creation_count.clone();
1164 move || {
1165 count.fetch_add(1, Ordering::Relaxed);
1166 NeverConnects
1167 }
1168 });
1169
1170 let settings = [
1172 HttpConnectorSettings::builder()
1173 .connect_timeout(Duration::from_secs(3))
1174 .build(),
1175 HttpConnectorSettings::builder()
1176 .read_timeout(Duration::from_secs(3))
1177 .build(),
1178 HttpConnectorSettings::builder()
1179 .connect_timeout(Duration::from_secs(3))
1180 .read_timeout(Duration::from_secs(3))
1181 .build(),
1182 HttpConnectorSettings::builder()
1183 .connect_timeout(Duration::from_secs(5))
1184 .read_timeout(Duration::from_secs(3))
1185 .build(),
1186 ];
1187
1188 let components = RuntimeComponentsBuilder::for_tests()
1190 .with_time_source(Some(SystemTimeSource::new()))
1191 .build()
1192 .unwrap();
1193 let mut handles = Vec::new();
1194 for setting in &settings {
1195 for _ in 0..1000 {
1196 let client = http_client.clone();
1197 handles.push(tokio::spawn({
1198 let setting = setting.clone();
1199 let components = components.clone();
1200 async move {
1201 let _ = client.http_connector(&setting, &components);
1202 }
1203 }));
1204 }
1205 }
1206 for handle in handles {
1207 handle.await.unwrap();
1208 }
1209
1210 assert_eq!(4, creation_count.load(Ordering::Relaxed));
1212 }
1213
1214 #[tokio::test]
1215 async fn hyper_io_error() {
1216 let connector = TestConnection {
1217 inner: HangupStream,
1218 };
1219 let adapter = Connector::builder().wrap_connector(connector).adapter;
1220 let err = adapter
1221 .call(HttpRequest::get("https://socket-hangup.com").unwrap())
1222 .await
1223 .expect_err("socket hangup");
1224 assert!(err.is_io(), "unexpected error type: {:?}", err);
1225 }
1226
1227 #[derive(Clone)]
1229 struct HangupStream;
1230
1231 impl Connection for HangupStream {
1232 fn connected(&self) -> Connected {
1233 Connected::new()
1234 }
1235 }
1236
1237 impl Read for HangupStream {
1238 fn poll_read(
1239 self: Pin<&mut Self>,
1240 _cx: &mut Context<'_>,
1241 _buf: ReadBufCursor<'_>,
1242 ) -> Poll<std::io::Result<()>> {
1243 Poll::Ready(Err(Error::new(
1244 ErrorKind::ConnectionReset,
1245 "connection reset",
1246 )))
1247 }
1248 }
1249
1250 impl Write for HangupStream {
1251 fn poll_write(
1252 self: Pin<&mut Self>,
1253 _cx: &mut Context<'_>,
1254 _buf: &[u8],
1255 ) -> Poll<Result<usize, Error>> {
1256 Poll::Pending
1257 }
1258
1259 fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1260 Poll::Pending
1261 }
1262
1263 fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1264 Poll::Pending
1265 }
1266 }
1267
1268 #[derive(Clone)]
1269 struct TestConnection<T> {
1270 inner: T,
1271 }
1272
1273 impl<T> tower::Service<Uri> for TestConnection<T>
1274 where
1275 T: Clone + Connection,
1276 {
1277 type Response = T;
1278 type Error = BoxError;
1279 type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
1280
1281 fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
1282 Poll::Ready(Ok(()))
1283 }
1284
1285 fn call(&mut self, _req: Uri) -> Self::Future {
1286 std::future::ready(Ok(self.inner.clone()))
1287 }
1288 }
1289
1290 #[tokio::test]
1291 async fn http_connect_timeout_works() {
1292 let tcp_connector = NeverConnects::default();
1293 let connector_settings = HttpConnectorSettings::builder()
1294 .connect_timeout(Duration::from_secs(1))
1295 .build();
1296 let hyper = Connector::builder()
1297 .connector_settings(connector_settings)
1298 .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1299 .wrap_connector(tcp_connector)
1300 .adapter;
1301 let now = tokio::time::Instant::now();
1302 tokio::time::pause();
1303 let resp = hyper
1304 .call(HttpRequest::get("https://static-uri.com").unwrap())
1305 .await
1306 .unwrap_err();
1307 assert!(
1308 resp.is_timeout(),
1309 "expected resp.is_timeout() to be true but it was false, resp == {:?}",
1310 resp
1311 );
1312 let message = DisplayErrorContext(&resp).to_string();
1313 let expected = "timeout: client error (Connect): HTTP connect timeout occurred after 1s";
1314 assert!(
1315 message.contains(expected),
1316 "expected '{message}' to contain '{expected}'"
1317 );
1318 assert_elapsed!(now, Duration::from_secs(1));
1319 }
1320
1321 #[tokio::test]
1322 async fn http_read_timeout_works() {
1323 let tcp_connector = crate::client::timeout::test::NeverReplies;
1324 let connector_settings = HttpConnectorSettings::builder()
1325 .connect_timeout(Duration::from_secs(1))
1326 .read_timeout(Duration::from_secs(2))
1327 .build();
1328 let hyper = Connector::builder()
1329 .connector_settings(connector_settings)
1330 .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1331 .wrap_connector(tcp_connector)
1332 .adapter;
1333 let now = tokio::time::Instant::now();
1334 tokio::time::pause();
1335 let err = hyper
1336 .call(HttpRequest::get("https://fake-uri.com").unwrap())
1337 .await
1338 .unwrap_err();
1339 assert!(
1340 err.is_timeout(),
1341 "expected err.is_timeout() to be true but it was false, err == {err:?}",
1342 );
1343 let message = format!("{}", DisplayErrorContext(&err));
1344 let expected = "timeout: HTTP read timeout occurred after 2s";
1345 assert!(
1346 message.contains(expected),
1347 "expected '{message}' to contain '{expected}'"
1348 );
1349 assert_elapsed!(now, Duration::from_secs(2));
1350 }
1351
1352 #[cfg(not(windows))]
1353 #[tokio::test]
1354 async fn connection_refused_works() {
1355 use crate::client::dns::HyperUtilResolver;
1356 use aws_smithy_runtime_api::client::dns::{DnsFuture, ResolveDns};
1357 use std::net::{IpAddr, Ipv4Addr};
1358
1359 #[derive(Debug, Clone, Default)]
1360 struct TestResolver;
1361 impl ResolveDns for TestResolver {
1362 fn resolve_dns<'a>(&'a self, _name: &'a str) -> DnsFuture<'a> {
1363 let localhost_v4 = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
1364 DnsFuture::ready(Ok(vec![localhost_v4]))
1365 }
1366 }
1367
1368 let connector_settings = HttpConnectorSettings::builder()
1369 .connect_timeout(Duration::from_secs(20))
1370 .build();
1371
1372 let resolver = HyperUtilResolver {
1373 resolver: TestResolver,
1374 };
1375 let connector = Connector::builder().base_connector_with_resolver(resolver);
1376
1377 let hyper = Connector::builder()
1378 .connector_settings(connector_settings)
1379 .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1380 .wrap_connector(connector)
1381 .adapter;
1382
1383 let resp = hyper
1384 .call(HttpRequest::get("http://static-uri:50227.com").unwrap())
1385 .await
1386 .unwrap_err();
1387 assert!(
1388 resp.is_io(),
1389 "expected resp.is_io() to be true but it was false, resp == {:?}",
1390 resp
1391 );
1392 let message = DisplayErrorContext(&resp).to_string();
1393 let expected = "Connection refused";
1394 assert!(
1395 message.contains(expected),
1396 "expected '{message}' to contain '{expected}'"
1397 );
1398 }
1399
1400 #[cfg(feature = "s2n-tls")]
1401 #[tokio::test]
1402 async fn s2n_tls_provider() {
1403 let client = Builder::new()
1405 .tls_provider(tls::Provider::S2nTls)
1406 .build_https();
1407 let connector_settings = HttpConnectorSettings::builder().build();
1408
1409 let runtime_components = RuntimeComponentsBuilder::for_tests()
1412 .with_time_source(Some(SystemTimeSource::new()))
1413 .build()
1414 .unwrap();
1415
1416 let connector = client.http_connector(&connector_settings, &runtime_components);
1417
1418 let error = connector
1423 .call(HttpRequest::get("notascheme://amazon.com").unwrap())
1424 .await
1425 .unwrap_err();
1426 let error = error.into_source();
1427 let s2n_error = error
1428 .source()
1429 .unwrap()
1430 .downcast_ref::<s2n_tls_hyper::error::Error>()
1431 .unwrap();
1432 assert!(matches!(
1433 s2n_error,
1434 s2n_tls_hyper::error::Error::InvalidScheme
1435 ));
1436 }
1437
1438 #[test]
1440 fn connector_metadata_identifies_hyper_1x() {
1441 let client = crate::Builder::new().build_http();
1442 let metadata = client
1443 .connector_metadata()
1444 .expect("connector metadata should be present");
1445 assert_eq!(metadata.name(), Cow::Borrowed("hyper"));
1446 assert_eq!(metadata.version(), Some(Cow::Borrowed("1.x")));
1447 }
1448}