1use rama_core::rt::Executor;
2
3use super::{
4 HttpConnectRequestAdapter, HttpConnector, HttpPooledConnector, HttpPooledConnectorConfig,
5};
6use crate::{
7 Layer, Service,
8 dns::client::{
9 DnsConnector, DnsConnectorLayer, GlobalDnsResolver, resolver::DnsAddressResolver,
10 },
11 error::BoxError,
12 extensions::ExtensionsRef,
13 http::{
14 Request, StreamingBody, client::proxy::layer::HttpProxyConnector,
15 layer::version_adapter::RequestVersionAdapter,
16 },
17 net::client::{
18 ConnectRequest, ConnectionError, ConnectorService, EstablishedClientConnection,
19 MapEstablishedConnection, ProxyRouteFailureCache, ProxyRouteFailureCacheConnector,
20 ProxyRoutesConnector, pool::PooledConnector,
21 },
22 service::BoxService,
23 tcp::client::service::TcpConnector,
24};
25use rama_http::layer::{
26 alt_svc::{AltSvc, AltSvcCache, AltSvcLayer},
27 http_service::HttpServiceConnector,
28};
29use rama_utils::macros::generate_set_and_with;
30#[cfg(feature = "boring")]
31use std::sync::Arc;
32use std::time::Duration;
33
34#[cfg(feature = "boring")]
35use crate::tls::boring::client as boring_client;
36
37#[cfg(feature = "rustls")]
38use crate::tls::rustls::client as rustls_client;
39#[cfg(any(feature = "rustls", feature = "boring"))]
40use {
41 crate::http::conn::FallbackHttpVersion, rama_core::layer::AddInputExtension,
42 rama_tls::client::TlsClientConfig,
43};
44
45#[cfg(feature = "socks5")]
46use crate::{http::client::proxy_connector::ProxyConnector, proxy::socks5::Socks5ProxyConnector};
47
48#[derive(Default)]
50pub struct EasyHttpConnectorBuilder<C = (), S = (), D = ()> {
51 connector: C,
52 stage: S,
53 dns: D,
55}
56
57#[non_exhaustive]
58#[derive(Debug, Default)]
59pub struct TransportStage;
60
61#[non_exhaustive]
62#[derive(Debug, Default)]
63pub struct DnsStage;
64
65#[non_exhaustive]
66#[derive(Debug, Default)]
67pub struct ProxyTunnelStage<const TLS_PROXY: bool = true>;
68
69#[non_exhaustive]
70#[derive(Debug, Default)]
71pub struct ProxyStage<const PROXY: bool = true>;
72
73#[non_exhaustive]
74#[derive(Debug, Default)]
75pub struct TlsStage<const PROXY: bool = true>;
76
77#[non_exhaustive]
78#[derive(Debug, Default)]
79pub struct HttpStage<const PROXY: bool = true>;
80
81#[non_exhaustive]
82#[derive(Debug, Default)]
83pub struct ProxyRouteFailureCacheStage;
84
85#[non_exhaustive]
86#[derive(Debug, Default)]
87pub struct PoolStage;
88
89impl EasyHttpConnectorBuilder {
90 #[must_use]
91 pub fn new() -> Self {
92 Self::default()
93 }
94
95 #[must_use]
96 pub fn with_default_transport_connector(
97 self,
98 ) -> EasyHttpConnectorBuilder<TcpConnector, TransportStage> {
99 let connector = TcpConnector::default();
100 EasyHttpConnectorBuilder {
101 dns: self.dns,
102 connector,
103 stage: Default::default(),
104 }
105 }
106
107 pub fn with_custom_transport_connector<C>(
109 self,
110 connector: C,
111 ) -> EasyHttpConnectorBuilder<C, TransportStage> {
112 EasyHttpConnectorBuilder {
113 dns: self.dns,
114 connector,
115 stage: Default::default(),
116 }
117 }
118}
119
120impl<T, Stage, D> EasyHttpConnectorBuilder<T, Stage, D> {
121 pub fn with_custom_connector<L>(
126 self,
127 connector_layer: L,
128 ) -> EasyHttpConnectorBuilder<L::Service, Stage, D>
129 where
130 L: Layer<T>,
131 {
132 self.map_connector(|c| connector_layer.into_layer(c))
133 }
134
135 pub fn map_connector<T2>(
140 self,
141 map_fn: impl FnOnce(T) -> T2,
142 ) -> EasyHttpConnectorBuilder<T2, Stage, D> {
143 let connector = map_fn(self.connector);
144 EasyHttpConnectorBuilder {
145 dns: self.dns,
146 connector,
147 stage: self.stage,
148 }
149 }
150}
151
152impl<T> EasyHttpConnectorBuilder<T, TransportStage> {
153 pub fn with_default_dns_connector(
155 self,
156 ) -> EasyHttpConnectorBuilder<DnsConnector<T>, DnsStage, DnsConnectorLayer> {
157 self.with_dns_address_resolver(GlobalDnsResolver::new())
158 }
159
160 pub fn with_dns_address_resolver<R: DnsAddressResolver + Clone>(
162 self,
163 resolver: R,
164 ) -> EasyHttpConnectorBuilder<DnsConnector<T, R>, DnsStage, DnsConnectorLayer<R>> {
165 self.with_dns_connector(DnsConnectorLayer::with_resolver(resolver))
166 }
167
168 pub fn without_dns_connector(self) -> EasyHttpConnectorBuilder<T, DnsStage> {
170 self.with_dns_connector(())
171 }
172
173 pub fn with_dns_connector<L: Layer<T>>(
180 self,
181 layer: L,
182 ) -> EasyHttpConnectorBuilder<L::Service, DnsStage, L> {
183 let connector = layer.layer(self.connector);
184 EasyHttpConnectorBuilder {
185 connector,
186 stage: DnsStage,
187 dns: layer,
188 }
189 }
190}
191
192impl<T, D> EasyHttpConnectorBuilder<T, DnsStage, D> {
193 pub fn with_custom_tls_proxy_connector<L>(
200 self,
201 connector_layer: L,
202 ) -> EasyHttpConnectorBuilder<L::Service, ProxyTunnelStage<true>, D>
203 where
204 L: Layer<T>,
205 {
206 let connector = connector_layer.into_layer(self.connector);
207 EasyHttpConnectorBuilder {
208 dns: self.dns,
209 connector,
210 stage: Default::default(),
211 }
212 }
213
214 #[cfg(feature = "boring")]
215 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
216 pub fn with_tls_proxy_support_using_boringssl(
222 self,
223 ) -> EasyHttpConnectorBuilder<
224 boring_client::TlsConnector<T, boring_client::ConnectorKindTunnel>,
225 ProxyTunnelStage<true>,
226 D,
227 > {
228 let connector = boring_client::TlsConnector::tunnel(self.connector, None);
229 EasyHttpConnectorBuilder {
230 dns: self.dns,
231 connector,
232 stage: Default::default(),
233 }
234 }
235
236 #[cfg(feature = "boring")]
237 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
238 pub fn with_tls_proxy_support_using_boringssl_config(
244 self,
245 config: TlsClientConfig,
246 ) -> EasyHttpConnectorBuilder<
247 boring_client::TlsConnector<T, boring_client::ConnectorKindTunnel>,
248 ProxyTunnelStage<true>,
249 D,
250 > {
251 let connector =
252 boring_client::TlsConnector::tunnel(self.connector, None).with_base_config(config);
253 EasyHttpConnectorBuilder {
254 dns: self.dns,
255 connector,
256 stage: Default::default(),
257 }
258 }
259
260 #[cfg(feature = "rustls")]
261 #[cfg_attr(docsrs, doc(cfg(feature = "rustls")))]
262 pub fn with_tls_proxy_support_using_rustls(
268 self,
269 ) -> EasyHttpConnectorBuilder<
270 rustls_client::TlsConnector<T, rustls_client::ConnectorKindTunnel>,
271 ProxyTunnelStage<true>,
272 D,
273 > {
274 let connector = rustls_client::TlsConnector::tunnel(self.connector, None);
275
276 EasyHttpConnectorBuilder {
277 dns: self.dns,
278 connector,
279 stage: Default::default(),
280 }
281 }
282
283 #[cfg(feature = "rustls")]
284 #[cfg_attr(docsrs, doc(cfg(feature = "rustls")))]
285 pub fn with_tls_proxy_support_using_rustls_config(
291 self,
292 config: TlsClientConfig,
293 ) -> EasyHttpConnectorBuilder<
294 rustls_client::TlsConnector<T, rustls_client::ConnectorKindTunnel>,
295 ProxyTunnelStage<true>,
296 D,
297 > {
298 let connector =
299 rustls_client::TlsConnector::tunnel(self.connector, None).with_base_config(config);
300
301 EasyHttpConnectorBuilder {
302 dns: self.dns,
303 connector,
304 stage: Default::default(),
305 }
306 }
307
308 pub fn without_tls_proxy_support(
314 self,
315 ) -> EasyHttpConnectorBuilder<T, ProxyTunnelStage<false>, D> {
316 EasyHttpConnectorBuilder {
317 dns: self.dns,
318 connector: self.connector,
319 stage: Default::default(),
320 }
321 }
322}
323
324impl<T, D, const TLS_PROXY: bool> EasyHttpConnectorBuilder<T, ProxyTunnelStage<TLS_PROXY>, D> {
325 pub fn with_custom_proxy_connector<L>(
327 self,
328 connector_layer: L,
329 ) -> EasyHttpConnectorBuilder<L::Service, ProxyStage<true>, D>
330 where
331 L: Layer<T>,
332 {
333 let connector = connector_layer.into_layer(self.connector);
334 EasyHttpConnectorBuilder {
335 dns: self.dns,
336 connector,
337 stage: Default::default(),
338 }
339 }
340
341 #[cfg(not(feature = "socks5"))]
342 pub fn with_proxy_support(
352 self,
353 ) -> EasyHttpConnectorBuilder<HttpProxyConnector<T>, ProxyStage<true>, D> {
354 self.with_http_proxy_support()
355 }
356
357 pub fn with_http_proxy_support(
365 self,
366 ) -> EasyHttpConnectorBuilder<HttpProxyConnector<T>, ProxyStage<true>, D> {
367 let connector =
368 HttpProxyConnector::optional(self.connector).with_tls_proxy_support(TLS_PROXY);
369
370 EasyHttpConnectorBuilder {
371 dns: self.dns,
372 connector,
373 stage: Default::default(),
374 }
375 }
376
377 #[cfg(feature = "socks5")]
378 #[cfg_attr(docsrs, doc(cfg(feature = "socks5")))]
379 pub fn with_socks5_proxy_support(
383 self,
384 ) -> EasyHttpConnectorBuilder<Socks5ProxyConnector<T>, ProxyStage<true>, D> {
385 let connector = Socks5ProxyConnector::optional(self.connector);
386
387 EasyHttpConnectorBuilder {
388 dns: self.dns,
389 connector,
390 stage: Default::default(),
391 }
392 }
393
394 pub fn without_proxy_support(self) -> EasyHttpConnectorBuilder<T, ProxyStage<false>, D> {
396 EasyHttpConnectorBuilder {
397 dns: self.dns,
398 connector: self.connector,
399 stage: Default::default(),
400 }
401 }
402}
403
404impl<T: Clone, D, const TLS_PROXY: bool>
405 EasyHttpConnectorBuilder<T, ProxyTunnelStage<TLS_PROXY>, D>
406{
407 #[cfg(feature = "socks5")]
408 #[cfg_attr(docsrs, doc(cfg(feature = "socks5")))]
409 pub fn with_proxy_support(
417 self,
418 ) -> EasyHttpConnectorBuilder<ProxyConnector<T>, ProxyStage<true>, D> {
419 use rama_http_backend::client::proxy::layer::HttpProxyConnectorLayer;
420 use rama_socks5::Socks5ProxyConnectorLayer;
421
422 let connector = ProxyConnector::optional(
423 self.connector,
424 Socks5ProxyConnectorLayer::required(),
425 HttpProxyConnectorLayer::required().with_tls_proxy_support(TLS_PROXY),
426 );
427
428 EasyHttpConnectorBuilder {
429 dns: self.dns,
430 connector,
431 stage: Default::default(),
432 }
433 }
434}
435
436impl<T, D, const PROXY: bool> EasyHttpConnectorBuilder<T, ProxyStage<PROXY>, D> {
437 pub fn with_custom_tls_connector<L>(
447 self,
448 connector_layer: L,
449 ) -> EasyHttpConnectorBuilder<L::Service, TlsStage<PROXY>, D>
450 where
451 L: Layer<T>,
452 {
453 let connector = connector_layer.into_layer(self.connector);
454
455 EasyHttpConnectorBuilder {
456 dns: self.dns,
457 connector,
458 stage: Default::default(),
459 }
460 }
461
462 #[cfg(feature = "boring")]
463 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
464 pub fn with_tls_support_using_boringssl(
469 self,
470 config: TlsClientConfig,
471 ) -> EasyHttpConnectorBuilder<boring_client::TlsConnector<T>, TlsStage<PROXY>, D> {
472 let connector = boring_client::TlsConnector::auto(self.connector).with_base_config(config);
473
474 EasyHttpConnectorBuilder {
475 dns: self.dns,
476 connector,
477 stage: Default::default(),
478 }
479 }
480
481 #[cfg(feature = "boring")]
482 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
483 pub fn with_tls_support_using_boringssl_and_default_http_version(
493 self,
494 config: TlsClientConfig,
495 default_http_version: rama_http::Version,
496 ) -> EasyHttpConnectorBuilder<
497 AddInputExtension<boring_client::TlsConnector<T>, FallbackHttpVersion>,
498 TlsStage<PROXY>,
499 D,
500 > {
501 let connector = boring_client::TlsConnector::auto(self.connector).with_base_config(config);
502 let connector =
503 AddInputExtension::new(connector, FallbackHttpVersion(default_http_version))
504 .with_overwrite(false);
505
506 EasyHttpConnectorBuilder {
507 dns: self.dns,
508 connector,
509 stage: Default::default(),
510 }
511 }
512
513 #[cfg(feature = "rustls")]
514 #[cfg_attr(docsrs, doc(cfg(feature = "rustls")))]
515 pub fn with_tls_support_using_rustls(
520 self,
521 config: TlsClientConfig,
522 ) -> EasyHttpConnectorBuilder<rustls_client::TlsConnector<T>, TlsStage<PROXY>, D> {
523 let connector = rustls_client::TlsConnector::auto(self.connector).with_base_config(config);
524
525 EasyHttpConnectorBuilder {
526 dns: self.dns,
527 connector,
528 stage: Default::default(),
529 }
530 }
531
532 #[cfg(feature = "rustls")]
533 #[cfg_attr(docsrs, doc(cfg(feature = "rustls")))]
534 pub fn with_tls_support_using_rustls_and_default_http_version(
544 self,
545 config: TlsClientConfig,
546 default_http_version: rama_http::Version,
547 ) -> EasyHttpConnectorBuilder<
548 AddInputExtension<rustls_client::TlsConnector<T>, FallbackHttpVersion>,
549 TlsStage<PROXY>,
550 D,
551 > {
552 let connector = rustls_client::TlsConnector::auto(self.connector).with_base_config(config);
553 let connector =
554 AddInputExtension::new(connector, FallbackHttpVersion(default_http_version))
555 .with_overwrite(false);
556
557 EasyHttpConnectorBuilder {
558 dns: self.dns,
559 connector,
560 stage: Default::default(),
561 }
562 }
563
564 pub fn without_tls_support(self) -> EasyHttpConnectorBuilder<T, TlsStage<PROXY>, D> {
566 EasyHttpConnectorBuilder {
567 dns: self.dns,
568 connector: self.connector,
569 stage: Default::default(),
570 }
571 }
572}
573
574#[cfg(feature = "boring")]
575impl<T, D, const PROXY: bool>
576 EasyHttpConnectorBuilder<boring_client::TlsConnector<T>, TlsStage<PROXY>, D>
577{
578 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
579 #[must_use]
585 pub fn with_tls_session_store(
586 mut self,
587 store: Arc<dyn boring_client::TlsClientSessionStore>,
588 ) -> Self {
589 self.connector.set_session_store(store);
590 self
591 }
592}
593
594impl<T, D, const PROXY: bool> EasyHttpConnectorBuilder<T, TlsStage<PROXY>, D> {
595 pub fn with_default_http_connector<Body>(
597 self,
598 exec: Executor,
599 ) -> EasyHttpConnectorBuilder<HttpConnector<T, Body>, HttpStage<PROXY>, D> {
600 let connector = HttpConnector::new(self.connector, exec);
601
602 EasyHttpConnectorBuilder {
603 dns: self.dns,
604 connector,
605 stage: HttpStage,
606 }
607 }
608
609 pub fn with_custom_http_connector<L>(
611 self,
612 connector_layer: L,
613 ) -> EasyHttpConnectorBuilder<L::Service, HttpStage<PROXY>, D>
614 where
615 L: Layer<T>,
616 {
617 let connector = connector_layer.into_layer(self.connector);
618
619 EasyHttpConnectorBuilder {
620 dns: self.dns,
621 connector,
622 stage: HttpStage,
623 }
624 }
625}
626
627impl<T, Body, D, const PROXY: bool>
628 EasyHttpConnectorBuilder<HttpConnector<T, Body>, HttpStage<PROXY>, D>
629{
630 pub fn with_http3_connector<C>(
636 self,
637 connector: C,
638 ) -> EasyHttpConnectorBuilder<
639 HttpConnector<super::HttpTransportConnector<T, C>, Body>,
640 HttpStage<PROXY>,
641 D,
642 > {
643 EasyHttpConnectorBuilder {
644 dns: self.dns,
645 connector: self.connector.with_http3_connector(connector),
646 stage: self.stage,
647 }
648 }
649
650 pub fn with_http3_support(
656 self,
657 connector: super::Http3Connector,
658 ) -> EasyHttpConnectorBuilder<
659 HttpConnector<super::HttpTransportConnector<T, D::Service>, Body>,
660 HttpStage<PROXY>,
661 D,
662 >
663 where
664 D: Layer<super::Http3Connector>,
665 {
666 let connector = self.dns.layer(connector);
667 self.with_http3_connector(connector)
668 }
669}
670
671type DefaultHttpConnector<T> = RequestVersionAdapter<
672 HttpConnectRequestAdapter<
673 MapEstablishedConnection<
674 HttpServiceConnector<ProxyRoutesConnector<T>>,
675 AltSvcConnectionLayer,
676 >,
677 >,
678>;
679
680#[doc(hidden)]
685#[derive(Clone, Debug)]
686pub struct AltSvcConnectionLayer {
687 cache: Option<AltSvcCache>,
688}
689
690impl<S> Layer<S> for AltSvcConnectionLayer {
691 type Service = AltSvc<S>;
692
693 fn layer(&self, inner: S) -> Self::Service {
694 match &self.cache {
695 Some(cache) => AltSvcLayer::new(cache.clone()).layer(inner),
696 None => AltSvc::passthrough(inner),
697 }
698 }
699}
700
701type ConfiguredConnectionBuilder<T> = EasyHttpConnectorBuilder<DefaultHttpConnector<T>, PoolStage>;
702
703type ConfiguredConnectionPoolBuilder<T> =
704 EasyHttpConnectorBuilder<DefaultHttpConnector<HttpPooledConnector<T>>, PoolStage>;
705
706type ErasedConnector<C> =
707 BoxService<ConnectRequest, EstablishedClientConnection<C, ConnectRequest>, ConnectionError>;
708
709type DefaultConnectionBuilder<C> =
710 ConfiguredConnectionBuilder<ProxyRouteFailureCacheConnector<ErasedConnector<C>>>;
711
712type DefaultConnectionPoolBuilder<C> =
713 ConfiguredConnectionPoolBuilder<ProxyRouteFailureCacheConnector<ErasedConnector<C>>>;
714
715struct ConnectorServiceAdapter<T>(T);
720
721impl<T> Service<ConnectRequest> for ConnectorServiceAdapter<T>
722where
723 T: ConnectorService<ConnectRequest>,
724{
725 type Output = EstablishedClientConnection<T::Connection, ConnectRequest>;
726 type Error = ConnectionError;
727
728 fn serve(
729 &self,
730 input: ConnectRequest,
731 ) -> impl Future<Output = Result<Self::Output, Self::Error>> + Send + '_ {
732 self.0.connect(input)
733 }
734}
735
736fn erase_connector<T>(connector: T) -> ErasedConnector<T::Connection>
737where
738 T: ConnectorService<ConnectRequest>,
739{
740 ConnectorServiceAdapter(connector).boxed()
741}
742
743fn finalize_http_connector<T>(connector: T) -> DefaultHttpConnector<T> {
744 let connector = ProxyRoutesConnector::new(connector);
745 let cache = AltSvcCache::default();
746 let connector = HttpServiceConnector::new(connector).with_cache(cache.clone());
747 adapt_http_service_connector(connector, Some(cache))
748}
749
750fn adapt_http_service_connector<T>(
751 connector: HttpServiceConnector<ProxyRoutesConnector<T>>,
752 cache: Option<AltSvcCache>,
753) -> DefaultHttpConnector<T> {
754 let connector = MapEstablishedConnection::new(connector, AltSvcConnectionLayer { cache });
755 RequestVersionAdapter::new(HttpConnectRequestAdapter::new(connector))
756}
757
758impl<T> EasyHttpConnectorBuilder<DefaultHttpConnector<T>, PoolStage> {
759 #[must_use]
761 pub fn without_alt_svc(self) -> Self {
762 self.maybe_with_alt_svc_cache(None)
763 }
764
765 generate_set_and_with! {
766 pub fn alt_svc_cache(mut self, cache: Option<AltSvcCache>) -> Self {
768 let connector = self.connector.get_mut().get_mut();
769 connector.get_mut().maybe_set_cache(cache.clone());
770 connector.layer_mut().cache = cache;
771 self
772 }
773 }
774}
775
776fn finish_without_connection_pool<T, Stage, D>(
777 builder: EasyHttpConnectorBuilder<T, Stage, D>,
778) -> ConfiguredConnectionBuilder<T>
779where
780 T: ConnectorService<ConnectRequest>,
781{
782 EasyHttpConnectorBuilder {
783 dns: (),
784 connector: finalize_http_connector(builder.connector),
785 stage: Default::default(),
786 }
787}
788
789fn finish_with_connection_pool<T, Stage, D>(
790 builder: EasyHttpConnectorBuilder<T, Stage, D>,
791 config: HttpPooledConnectorConfig,
792) -> ConfiguredConnectionPoolBuilder<T>
793where
794 T: ConnectorService<ConnectRequest>,
795{
796 EasyHttpConnectorBuilder {
797 dns: (),
798 connector: finalize_http_connector(config.build_connector(builder.connector)),
799 stage: Default::default(),
800 }
801}
802
803fn finish_with_custom_connection_pool<T, Stage, D, P, R>(
804 builder: EasyHttpConnectorBuilder<T, Stage, D>,
805 pool: P,
806 req_to_conn_id: R,
807 wait_for_pool_timeout: Option<Duration>,
808) -> EasyHttpConnectorBuilder<PooledConnector<T, P, R>, PoolStage> {
809 let connector = PooledConnector::new(builder.connector, pool, req_to_conn_id)
810 .maybe_with_wait_for_pool_timeout(wait_for_pool_timeout);
811 EasyHttpConnectorBuilder {
812 dns: (),
813 connector,
814 stage: Default::default(),
815 }
816}
817
818impl<T, D, const PROXY: bool> EasyHttpConnectorBuilder<T, HttpStage<PROXY>, D> {
819 #[must_use]
825 pub fn with_proxy_route_failure_cache(
826 self,
827 cache: ProxyRouteFailureCache,
828 ) -> EasyHttpConnectorBuilder<
829 ProxyRouteFailureCacheConnector<ErasedConnector<T::Connection>>,
830 ProxyRouteFailureCacheStage,
831 D,
832 >
833 where
834 T: ConnectorService<ConnectRequest>,
835 {
836 EasyHttpConnectorBuilder {
837 dns: self.dns,
838 connector: ProxyRouteFailureCacheConnector::new(erase_connector(self.connector), cache),
839 stage: ProxyRouteFailureCacheStage,
840 }
841 }
842
843 #[must_use]
845 pub fn without_proxy_route_failure_cache(
846 self,
847 ) -> EasyHttpConnectorBuilder<T, ProxyRouteFailureCacheStage, D> {
848 EasyHttpConnectorBuilder {
849 dns: self.dns,
850 connector: self.connector,
851 stage: ProxyRouteFailureCacheStage,
852 }
853 }
854}
855
856impl<T, D> EasyHttpConnectorBuilder<T, HttpStage<true>, D> {
857 pub fn without_connection_pool(self) -> DefaultConnectionBuilder<T::Connection>
863 where
864 T: ConnectorService<ConnectRequest>,
865 {
866 finish_without_connection_pool(
867 self.with_proxy_route_failure_cache(ProxyRouteFailureCache::default()),
868 )
869 }
870
871 pub fn with_connection_pool(
892 self,
893 config: HttpPooledConnectorConfig,
894 ) -> DefaultConnectionPoolBuilder<T::Connection>
895 where
896 T: ConnectorService<ConnectRequest>,
897 {
898 finish_with_connection_pool(
899 self.with_proxy_route_failure_cache(ProxyRouteFailureCache::default()),
900 config,
901 )
902 }
903
904 pub fn with_default_connection_pool(self) -> DefaultConnectionPoolBuilder<T::Connection>
907 where
908 T: ConnectorService<ConnectRequest>,
909 {
910 self.with_connection_pool(HttpPooledConnectorConfig::default())
911 }
912
913 pub fn with_custom_connection_pool<P, R>(
939 self,
940 pool: P,
941 req_to_conn_id: R,
942 wait_for_pool_timeout: Option<Duration>,
943 ) -> EasyHttpConnectorBuilder<
944 PooledConnector<ProxyRouteFailureCacheConnector<ErasedConnector<T::Connection>>, P, R>,
945 PoolStage,
946 >
947 where
948 T: ConnectorService<ConnectRequest>,
949 {
950 finish_with_custom_connection_pool(
951 self.with_proxy_route_failure_cache(ProxyRouteFailureCache::default()),
952 pool,
953 req_to_conn_id,
954 wait_for_pool_timeout,
955 )
956 }
957}
958
959impl<T, D> EasyHttpConnectorBuilder<T, HttpStage<false>, D> {
960 pub fn without_connection_pool(self) -> ConfiguredConnectionBuilder<T>
966 where
967 T: ConnectorService<ConnectRequest>,
968 {
969 finish_without_connection_pool(self)
970 }
971
972 pub fn with_connection_pool(
974 self,
975 config: HttpPooledConnectorConfig,
976 ) -> ConfiguredConnectionPoolBuilder<T>
977 where
978 T: ConnectorService<ConnectRequest>,
979 {
980 finish_with_connection_pool(self, config)
981 }
982
983 pub fn with_default_connection_pool(self) -> ConfiguredConnectionPoolBuilder<T>
986 where
987 T: ConnectorService<ConnectRequest>,
988 {
989 finish_with_connection_pool(self, HttpPooledConnectorConfig::default())
990 }
991
992 pub fn with_custom_connection_pool<P, R>(
999 self,
1000 pool: P,
1001 req_to_conn_id: R,
1002 wait_for_pool_timeout: Option<Duration>,
1003 ) -> EasyHttpConnectorBuilder<PooledConnector<T, P, R>, PoolStage> {
1004 finish_with_custom_connection_pool(self, pool, req_to_conn_id, wait_for_pool_timeout)
1005 }
1006}
1007
1008impl<T, D> EasyHttpConnectorBuilder<T, ProxyRouteFailureCacheStage, D> {
1009 pub fn without_connection_pool(self) -> ConfiguredConnectionBuilder<T>
1011 where
1012 T: ConnectorService<ConnectRequest>,
1013 {
1014 finish_without_connection_pool(self)
1015 }
1016
1017 pub fn with_connection_pool(
1019 self,
1020 config: HttpPooledConnectorConfig,
1021 ) -> ConfiguredConnectionPoolBuilder<T>
1022 where
1023 T: ConnectorService<ConnectRequest>,
1024 {
1025 finish_with_connection_pool(self, config)
1026 }
1027
1028 pub fn with_default_connection_pool(self) -> ConfiguredConnectionPoolBuilder<T>
1031 where
1032 T: ConnectorService<ConnectRequest>,
1033 {
1034 finish_with_connection_pool(self, HttpPooledConnectorConfig::default())
1035 }
1036
1037 pub fn with_custom_connection_pool<P, R>(
1050 self,
1051 pool: P,
1052 req_to_conn_id: R,
1053 wait_for_pool_timeout: Option<Duration>,
1054 ) -> EasyHttpConnectorBuilder<PooledConnector<T, P, R>, PoolStage> {
1055 finish_with_custom_connection_pool(self, pool, req_to_conn_id, wait_for_pool_timeout)
1056 }
1057}
1058
1059impl<T> EasyHttpConnectorBuilder<T, PoolStage> {
1060 pub fn build_client<Body, ModifiedBody, ConnResponse>(
1062 self,
1063 ) -> super::EasyHttpWebClient<Body, T::Output, ()>
1064 where
1065 Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
1066 ModifiedBody:
1067 StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
1068 T: Service<
1069 Request<Body>,
1070 Output = EstablishedClientConnection<ConnResponse, Request<ModifiedBody>>,
1071 Error: Into<BoxError>,
1072 >,
1073 ConnResponse: ExtensionsRef,
1074 {
1075 super::EasyHttpWebClient::new(self.connector)
1076 }
1077}
1078
1079impl<T, S, D> EasyHttpConnectorBuilder<T, S, D> {
1080 pub fn build_connector(self) -> T {
1082 self.connector
1083 }
1084}
1085
1086#[cfg(test)]
1087mod tests {
1088 use super::*;
1089 use crate::http::{Body, client::HttpConnIdentifier};
1090 use rama_net::{
1091 Protocol,
1092 address::HostWithPort,
1093 client::{
1094 ConnectRequest,
1095 pool::{ConnID as _, ReqToConnID},
1096 },
1097 };
1098 use rama_tls::client::{ServerVerifyMode, TlsPoolId, TlsServerVerify};
1099 use rama_utils::octets::kib;
1100
1101 use rama_core::layer::layer_fn;
1102 #[cfg(any(
1103 feature = "boring",
1104 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1105 ))]
1106 use {
1107 crate::http::client::Http3Connector,
1108 crate::quic::Endpoint,
1109 rama_core::futures::{Stream, stream},
1110 rama_http::layer::http_service::HttpServiceAttempt,
1111 rama_net::{
1112 address::{Domain, SocketAddress},
1113 client::ConnectionErrorKind,
1114 http::TargetHttpVersion as RequestedVersion,
1115 },
1116 std::{
1117 net::{Ipv4Addr, Ipv6Addr, SocketAddr},
1118 sync::{
1119 Arc,
1120 atomic::{AtomicUsize, Ordering},
1121 },
1122 },
1123 };
1124 #[cfg(feature = "tls")]
1125 use {
1126 crate::http::{
1127 HeaderMap, Version,
1128 conn::{HttpOrigin, SelectedHttpService},
1129 header::ALT_SVC,
1130 },
1131 rama_core::{ServiceInput, service::service_fn},
1132 rama_net::{http::TargetHttpVersion, tls::ApplicationProtocol},
1133 rama_tls::{
1134 ProtocolVersion,
1135 client::{NegotiatedTlsParameters, TlsServerAuthentication},
1136 },
1137 std::convert::Infallible,
1138 };
1139
1140 fn assert_future_budget(connector: &impl Service<Request>, name: &str) {
1141 let request = Request::builder()
1142 .uri("https://example.com/")
1143 .body(Body::empty())
1144 .unwrap();
1145 let future = connector.serve(request);
1146 let size = std::mem::size_of_val(&future);
1147 assert!(
1148 size <= kib(64),
1149 "{name} connector future is {size} bytes; connector error adapters must not duplicate nested future storage"
1150 );
1151 }
1152
1153 #[tokio::test]
1154 async fn default_connector_futures_stay_within_stack_budget() {
1155 let builder = || {
1156 EasyHttpConnectorBuilder::new()
1157 .with_default_transport_connector()
1158 .without_dns_connector()
1159 .without_tls_proxy_support()
1160 .without_proxy_support()
1161 };
1162 let plain = builder()
1163 .without_tls_support()
1164 .with_default_http_connector::<Body>(Executor::default())
1165 .with_default_connection_pool()
1166 .build_connector();
1167 assert_future_budget(&plain, "plain pooled");
1168 #[cfg(feature = "rustls")]
1169 {
1170 let pooled = builder()
1171 .with_tls_support_using_rustls(TlsClientConfig::new())
1172 .with_default_http_connector::<Body>(Executor::default())
1173 .with_default_connection_pool()
1174 .build_connector();
1175 assert_future_budget(&pooled, "rustls pooled");
1176 let unpooled = builder()
1177 .with_tls_support_using_rustls(TlsClientConfig::new())
1178 .with_default_http_connector::<Body>(Executor::default())
1179 .without_connection_pool()
1180 .build_connector();
1181 assert_future_budget(&unpooled, "rustls unpooled");
1182 }
1183 }
1184
1185 #[cfg(feature = "tls")]
1186 #[tokio::test]
1187 async fn custom_http_connector_discovers_h3_without_capability_configuration() {
1188 let cache = AltSvcCache::default();
1189 let origin =
1190 HttpOrigin::new(Protocol::HTTPS, HostWithPort::example_domain_https()).unwrap();
1191 let mut headers = HeaderMap::new();
1192 headers.insert(ALT_SVC, "h3=\":8443\"".parse().unwrap());
1193 cache.record(&origin, &headers, Duration::ZERO);
1194
1195 let builder = EasyHttpConnectorBuilder::new()
1196 .with_custom_transport_connector(())
1197 .without_dns_connector()
1198 .without_tls_proxy_support()
1199 .without_proxy_support()
1200 .without_tls_support()
1201 .with_custom_http_connector(layer_fn(|()| {
1202 service_fn(|input: ConnectRequest| async move {
1203 let version = input
1204 .extensions()
1205 .get_ref::<TargetHttpVersion>()
1206 .map_or(Version::HTTP_2, |version| version.0);
1207 let conn = ServiceInput::new(());
1208 conn.extensions.insert(TargetHttpVersion(version));
1209 conn.extensions
1210 .insert(TlsServerAuthentication(Some(input.authority.host.clone())));
1211 conn.extensions.insert(NegotiatedTlsParameters {
1212 protocol_version: ProtocolVersion::TLSv1_3,
1213 application_layer_protocol: Some(if version == Version::HTTP_3 {
1214 ApplicationProtocol::HTTP_3
1215 } else {
1216 ApplicationProtocol::HTTP_2
1217 }),
1218 peer_certificate_chain: None,
1219 server_name: None,
1220 resumed: None,
1221 algorithms: Default::default(),
1222 server_hello: None,
1223 });
1224 Ok::<_, Infallible>(EstablishedClientConnection { input, conn })
1225 })
1226 }))
1227 .without_connection_pool()
1228 .with_alt_svc_cache(cache.clone());
1229 let connector = builder.build_connector();
1230 for _ in 0..2 {
1231 let established = connector
1232 .serve(
1233 Request::builder()
1234 .uri("https://example.com/")
1235 .body(Body::empty())
1236 .unwrap(),
1237 )
1238 .await
1239 .unwrap();
1240 assert!(
1241 established
1242 .input
1243 .extensions()
1244 .contains::<SelectedHttpService>()
1245 );
1246 assert_eq!(established.input.version(), Version::HTTP_3);
1247 assert_eq!(
1248 established
1249 .input
1250 .extensions()
1251 .get_ref::<SelectedHttpService>()
1252 .unwrap()
1253 .candidate
1254 .target
1255 .port,
1256 8443
1257 );
1258 }
1259 }
1260
1261 #[test]
1262 fn custom_tls_proxy_layer_requires_no_builtin_provider() {
1263 EasyHttpConnectorBuilder::new()
1264 .with_custom_transport_connector(())
1265 .without_dns_connector()
1266 .with_custom_tls_proxy_connector(layer_fn(|inner| inner))
1267 .build_connector();
1268 }
1269
1270 #[test]
1271 fn custom_fixed_policy_pool_respects_explicit_override_identity() {
1272 let identifier = HttpConnIdentifier::default();
1273 let input = ConnectRequest::new(HostWithPort::example_domain_https());
1274 let fixed = identifier.id(&input).unwrap();
1275 assert!(fixed.is_reusable());
1276 let policy = TlsPoolId::builder()
1277 .with_verify(&TlsServerVerify(ServerVerifyMode::Disable))
1278 .build()
1279 .unwrap();
1280 input.extensions.insert(policy);
1281 let overridden = identifier.id(&input).unwrap();
1282 assert!(overridden.is_reusable());
1283 assert_ne!(fixed, overridden);
1284 assert_eq!(overridden, identifier.id(&input).unwrap());
1285 input.extensions.insert(TlsPoolId::non_reusable());
1286 assert!(!identifier.id(&input).unwrap().is_reusable());
1287 }
1288
1289 #[cfg(any(
1290 feature = "boring",
1291 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1292 ))]
1293 #[derive(Clone)]
1294 struct RecordingResolver {
1295 calls: Arc<AtomicUsize>,
1296 loopback: bool,
1297 }
1298
1299 #[cfg(any(
1300 feature = "boring",
1301 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1302 ))]
1303 impl DnsAddressResolver for RecordingResolver {
1304 type Error = Infallible;
1305
1306 fn lookup_ipv4(
1307 &self,
1308 domain: Domain,
1309 ) -> impl Stream<Item = Result<Ipv4Addr, Self::Error>> + Send + '_ {
1310 assert_eq!(domain.as_str(), "private.invalid");
1311 self.calls.fetch_add(1, Ordering::Relaxed);
1312 stream::iter(self.loopback.then_some(Ok(Ipv4Addr::LOCALHOST)))
1313 }
1314
1315 fn lookup_ipv6(
1316 &self,
1317 domain: Domain,
1318 ) -> impl Stream<Item = Result<Ipv6Addr, Self::Error>> + Send + '_ {
1319 assert_eq!(domain.as_str(), "private.invalid");
1320 self.calls.fetch_add(1, Ordering::Relaxed);
1321 stream::iter(self.loopback.then_some(Ok(Ipv6Addr::LOCALHOST)))
1322 }
1323 }
1324
1325 #[cfg(any(
1326 feature = "boring",
1327 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1328 ))]
1329 struct RecordingDnsLayer {
1330 resolver: RecordingResolver,
1331 applied: Arc<AtomicUsize>,
1332 }
1333
1334 #[cfg(any(
1335 feature = "boring",
1336 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1337 ))]
1338 impl<S> Layer<S> for RecordingDnsLayer {
1339 type Service = DnsConnector<S, RecordingResolver>;
1340
1341 fn layer(&self, inner: S) -> Self::Service {
1342 self.applied.fetch_add(1, Ordering::Relaxed);
1343 DnsConnector::with_resolver(inner, self.resolver.clone())
1344 }
1345 }
1346
1347 #[cfg(any(
1348 feature = "boring",
1349 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1350 ))]
1351 #[tokio::test]
1352 async fn built_in_quic_uses_the_configured_dns_resolver() {
1353 let calls = Arc::new(AtomicUsize::new(0));
1354 let applied = Arc::new(AtomicUsize::new(0));
1355 let executor = Executor::default();
1356 let h3 = Http3Connector::builder(executor.clone())
1357 .build()
1358 .await
1359 .unwrap();
1360 let connector = EasyHttpConnectorBuilder::new()
1361 .with_default_transport_connector()
1362 .with_dns_connector(RecordingDnsLayer {
1363 resolver: RecordingResolver {
1364 calls: calls.clone(),
1365 loopback: false,
1366 },
1367 applied: applied.clone(),
1368 })
1369 .without_tls_proxy_support()
1370 .without_proxy_support()
1371 .without_tls_support()
1372 .with_default_http_connector::<Body>(executor)
1373 .with_http3_support(h3)
1374 .build_connector();
1375 let input = ConnectRequest::new("private.invalid:443".parse().unwrap())
1376 .with_application_protocol(Protocol::HTTPS);
1377 input.extensions.insert(RequestedVersion(Version::HTTP_3));
1378 let error = connector.serve(input).await.err().unwrap();
1379 assert_eq!(error.kind(), ConnectionErrorKind::Unavailable);
1380 assert!(calls.load(Ordering::Relaxed) > 0);
1381 assert_eq!(
1382 applied.load(Ordering::Relaxed),
1383 2,
1384 "the same DNS layer must wrap both transports"
1385 );
1386 }
1387
1388 #[cfg(any(
1389 feature = "boring",
1390 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1391 ))]
1392 #[tokio::test]
1393 async fn ipv4_quic_endpoint_skips_ipv6_candidates_without_terminal_failure() {
1394 let executor = Executor::default();
1395 let endpoint = Endpoint::bind_client(executor.clone(), SocketAddress::local_ipv4(0))
1396 .await
1397 .unwrap();
1398 let blackhole = tokio::net::UdpSocket::bind(SocketAddr::from(SocketAddress::local_ipv4(0)))
1399 .await
1400 .unwrap();
1401 let connector = Http3Connector::builder(executor)
1402 .with_endpoint(endpoint.clone())
1403 .build()
1404 .await
1405 .unwrap();
1406 let calls = Arc::new(AtomicUsize::new(0));
1407 let connector = DnsConnector::with_resolver(
1408 connector,
1409 RecordingResolver {
1410 calls: calls.clone(),
1411 loopback: true,
1412 },
1413 );
1414 let input = ConnectRequest::new(
1415 format!("private.invalid:{}", blackhole.local_addr().unwrap().port())
1416 .parse()
1417 .unwrap(),
1418 )
1419 .with_application_protocol(Protocol::HTTPS);
1420 let attempt = Arc::new(HttpServiceAttempt::default());
1421 input.extensions.insert_arc(attempt.clone());
1422 tokio::time::timeout(Duration::from_millis(500), connector.serve(input))
1423 .await
1424 .unwrap_err();
1425 assert!(calls.load(Ordering::Relaxed) >= 2);
1426 assert!(
1427 !attempt.failed(),
1428 "a mismatched candidate family must not turn the IPv4 timeout into an authentication failure"
1429 );
1430 endpoint.close(0u32, b"test complete");
1431 }
1432}