1use rama_core::{error::extra::OpaqueError, graceful::WeakScope, layer::MapInput};
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 Body as ResponseBody, Request, Response, StreamingBody,
15 client::proxy::layer::HttpProxyConnector, 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 retry::{
29 ManagedPolicy, Retry, RetryBody, RetryError,
30 early_data::{is_early_data_rejected, permit_attempt, with_recovery},
31 managed::{PrepareBody, RetryRule},
32 },
33};
34use rama_utils::macros::generate_set_and_with;
35#[cfg(feature = "boring")]
36use std::sync::Arc;
37use std::time::Duration;
38
39#[cfg(feature = "boring")]
40use crate::tls::boring::client as boring_client;
41
42#[cfg(feature = "rustls")]
43use crate::tls::rustls::client as rustls_client;
44#[cfg(any(feature = "rustls", feature = "boring"))]
45use {
46 crate::http::conn::FallbackHttpVersion, rama_core::layer::AddInputExtension,
47 rama_tls::client::TlsClientConfig,
48};
49
50#[cfg(feature = "socks5")]
51use crate::{http::client::proxy_connector::ProxyConnector, proxy::socks5::Socks5ProxyConnector};
52
53#[cfg(all(feature = "masque", feature = "quic"))]
54use crate::{
55 proxy::masque::connect_udp::ConnectUdpConnector, quic::client::QuicEndpointConnector,
56 udp::proxy::UdpConnector,
57};
58
59#[derive(Default)]
61pub struct EasyHttpConnectorBuilder<C = (), S = (), D = ()> {
62 connector: C,
63 stage: S,
64 dns: D,
66}
67
68#[non_exhaustive]
69#[derive(Debug, Default)]
70pub struct TransportStage;
71
72#[non_exhaustive]
73#[derive(Debug, Default)]
74pub struct DnsStage;
75
76#[non_exhaustive]
77#[derive(Debug, Default)]
78pub struct ProxyTunnelStage<const TLS_PROXY: bool = true>;
79
80#[non_exhaustive]
81#[derive(Debug, Default)]
82pub struct ProxyStage<const PROXY: bool = true>;
83
84#[non_exhaustive]
85#[derive(Debug, Default)]
86pub struct TlsStage<const PROXY: bool = true>;
87
88#[non_exhaustive]
89#[derive(Debug, Default)]
90pub struct HttpStage<const PROXY: bool = true>;
91
92#[non_exhaustive]
93#[derive(Debug, Default)]
94pub struct ProxyRouteFailureCacheStage;
95
96#[non_exhaustive]
97#[derive(Debug, Default)]
98pub struct PoolStage;
99
100impl EasyHttpConnectorBuilder {
101 #[must_use]
102 pub fn new() -> Self {
103 Self::default()
104 }
105
106 #[must_use]
107 pub fn with_default_transport_connector(
108 self,
109 ) -> EasyHttpConnectorBuilder<TcpConnector, TransportStage> {
110 let connector = TcpConnector::default();
111 EasyHttpConnectorBuilder {
112 dns: self.dns,
113 connector,
114 stage: Default::default(),
115 }
116 }
117
118 pub fn with_custom_transport_connector<C>(
120 self,
121 connector: C,
122 ) -> EasyHttpConnectorBuilder<C, TransportStage> {
123 EasyHttpConnectorBuilder {
124 dns: self.dns,
125 connector,
126 stage: Default::default(),
127 }
128 }
129}
130
131impl<T, Stage, D> EasyHttpConnectorBuilder<T, Stage, D> {
132 pub fn with_custom_connector<L>(
137 self,
138 connector_layer: L,
139 ) -> EasyHttpConnectorBuilder<L::Service, Stage, D>
140 where
141 L: Layer<T>,
142 {
143 self.map_connector(|c| connector_layer.into_layer(c))
144 }
145
146 pub fn map_connector<T2>(
151 self,
152 map_fn: impl FnOnce(T) -> T2,
153 ) -> EasyHttpConnectorBuilder<T2, Stage, D> {
154 let connector = map_fn(self.connector);
155 EasyHttpConnectorBuilder {
156 dns: self.dns,
157 connector,
158 stage: self.stage,
159 }
160 }
161}
162
163impl<T> EasyHttpConnectorBuilder<T, TransportStage> {
164 pub fn with_default_dns_connector(
166 self,
167 ) -> EasyHttpConnectorBuilder<DnsConnector<T>, DnsStage, DnsConnectorLayer> {
168 self.with_dns_address_resolver(GlobalDnsResolver::new())
169 }
170
171 pub fn with_dns_address_resolver<R: DnsAddressResolver + Clone>(
173 self,
174 resolver: R,
175 ) -> EasyHttpConnectorBuilder<DnsConnector<T, R>, DnsStage, DnsConnectorLayer<R>> {
176 self.with_dns_connector(DnsConnectorLayer::with_resolver(resolver))
177 }
178
179 pub fn without_dns_connector(self) -> EasyHttpConnectorBuilder<T, DnsStage> {
181 self.with_dns_connector(())
182 }
183
184 pub fn with_dns_connector<L: Layer<T>>(
191 self,
192 layer: L,
193 ) -> EasyHttpConnectorBuilder<L::Service, DnsStage, L> {
194 let connector = layer.layer(self.connector);
195 EasyHttpConnectorBuilder {
196 connector,
197 stage: DnsStage,
198 dns: layer,
199 }
200 }
201}
202
203impl<T, D> EasyHttpConnectorBuilder<T, DnsStage, D> {
204 pub fn with_custom_tls_proxy_connector<L>(
211 self,
212 connector_layer: L,
213 ) -> EasyHttpConnectorBuilder<L::Service, ProxyTunnelStage<true>, D>
214 where
215 L: Layer<T>,
216 {
217 let connector = connector_layer.into_layer(self.connector);
218 EasyHttpConnectorBuilder {
219 dns: self.dns,
220 connector,
221 stage: Default::default(),
222 }
223 }
224
225 #[cfg(feature = "boring")]
226 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
227 pub fn with_tls_proxy_support_using_boringssl(
233 self,
234 ) -> EasyHttpConnectorBuilder<
235 boring_client::TlsConnector<T, boring_client::ConnectorKindTunnel>,
236 ProxyTunnelStage<true>,
237 D,
238 > {
239 let connector = boring_client::TlsConnector::tunnel(self.connector, None);
240 EasyHttpConnectorBuilder {
241 dns: self.dns,
242 connector,
243 stage: Default::default(),
244 }
245 }
246
247 #[cfg(feature = "boring")]
248 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
249 pub fn with_tls_proxy_support_using_boringssl_config(
255 self,
256 config: TlsClientConfig,
257 ) -> EasyHttpConnectorBuilder<
258 boring_client::TlsConnector<T, boring_client::ConnectorKindTunnel>,
259 ProxyTunnelStage<true>,
260 D,
261 > {
262 let connector =
263 boring_client::TlsConnector::tunnel(self.connector, None).with_base_config(config);
264 EasyHttpConnectorBuilder {
265 dns: self.dns,
266 connector,
267 stage: Default::default(),
268 }
269 }
270
271 #[cfg(feature = "rustls")]
272 #[cfg_attr(docsrs, doc(cfg(feature = "rustls")))]
273 pub fn with_tls_proxy_support_using_rustls(
279 self,
280 ) -> EasyHttpConnectorBuilder<
281 rustls_client::TlsConnector<T, rustls_client::ConnectorKindTunnel>,
282 ProxyTunnelStage<true>,
283 D,
284 > {
285 let connector = rustls_client::TlsConnector::tunnel(self.connector, None);
286
287 EasyHttpConnectorBuilder {
288 dns: self.dns,
289 connector,
290 stage: Default::default(),
291 }
292 }
293
294 #[cfg(feature = "rustls")]
295 #[cfg_attr(docsrs, doc(cfg(feature = "rustls")))]
296 pub fn with_tls_proxy_support_using_rustls_config(
302 self,
303 config: TlsClientConfig,
304 ) -> EasyHttpConnectorBuilder<
305 rustls_client::TlsConnector<T, rustls_client::ConnectorKindTunnel>,
306 ProxyTunnelStage<true>,
307 D,
308 > {
309 let connector =
310 rustls_client::TlsConnector::tunnel(self.connector, None).with_base_config(config);
311
312 EasyHttpConnectorBuilder {
313 dns: self.dns,
314 connector,
315 stage: Default::default(),
316 }
317 }
318
319 pub fn without_tls_proxy_support(
325 self,
326 ) -> EasyHttpConnectorBuilder<T, ProxyTunnelStage<false>, D> {
327 EasyHttpConnectorBuilder {
328 dns: self.dns,
329 connector: self.connector,
330 stage: Default::default(),
331 }
332 }
333}
334
335impl<T, D, const TLS_PROXY: bool> EasyHttpConnectorBuilder<T, ProxyTunnelStage<TLS_PROXY>, D> {
336 pub fn with_custom_proxy_connector<L>(
338 self,
339 connector_layer: L,
340 ) -> EasyHttpConnectorBuilder<L::Service, ProxyStage<true>, D>
341 where
342 L: Layer<T>,
343 {
344 let connector = connector_layer.into_layer(self.connector);
345 EasyHttpConnectorBuilder {
346 dns: self.dns,
347 connector,
348 stage: Default::default(),
349 }
350 }
351
352 #[cfg(not(feature = "socks5"))]
353 pub fn with_proxy_support(
363 self,
364 ) -> EasyHttpConnectorBuilder<HttpProxyConnector<T>, ProxyStage<true>, D> {
365 self.with_http_proxy_support()
366 }
367
368 pub fn with_http_proxy_support(
376 self,
377 ) -> EasyHttpConnectorBuilder<HttpProxyConnector<T>, ProxyStage<true>, D> {
378 let connector =
379 HttpProxyConnector::optional(self.connector).with_tls_proxy_support(TLS_PROXY);
380
381 EasyHttpConnectorBuilder {
382 dns: self.dns,
383 connector,
384 stage: Default::default(),
385 }
386 }
387
388 #[cfg(feature = "socks5")]
389 #[cfg_attr(docsrs, doc(cfg(feature = "socks5")))]
390 pub fn with_socks5_proxy_support(
394 self,
395 ) -> EasyHttpConnectorBuilder<Socks5ProxyConnector<T>, ProxyStage<true>, D> {
396 let connector = Socks5ProxyConnector::optional(self.connector);
397
398 EasyHttpConnectorBuilder {
399 dns: self.dns,
400 connector,
401 stage: Default::default(),
402 }
403 }
404
405 pub fn without_proxy_support(self) -> EasyHttpConnectorBuilder<T, ProxyStage<false>, D> {
407 EasyHttpConnectorBuilder {
408 dns: self.dns,
409 connector: self.connector,
410 stage: Default::default(),
411 }
412 }
413}
414
415impl<T: Clone, D, const TLS_PROXY: bool>
416 EasyHttpConnectorBuilder<T, ProxyTunnelStage<TLS_PROXY>, D>
417{
418 #[cfg(feature = "socks5")]
419 #[cfg_attr(docsrs, doc(cfg(feature = "socks5")))]
420 pub fn with_proxy_support(
428 self,
429 ) -> EasyHttpConnectorBuilder<ProxyConnector<T>, ProxyStage<true>, D> {
430 use rama_http_backend::client::proxy::layer::HttpProxyConnectorLayer;
431 use rama_socks5::Socks5ProxyConnectorLayer;
432
433 let connector = ProxyConnector::optional(
434 self.connector,
435 Socks5ProxyConnectorLayer::required(),
436 HttpProxyConnectorLayer::required().with_tls_proxy_support(TLS_PROXY),
437 );
438
439 EasyHttpConnectorBuilder {
440 dns: self.dns,
441 connector,
442 stage: Default::default(),
443 }
444 }
445}
446
447impl<T, D, const PROXY: bool> EasyHttpConnectorBuilder<T, ProxyStage<PROXY>, D> {
448 pub fn with_custom_tls_connector<L>(
458 self,
459 connector_layer: L,
460 ) -> EasyHttpConnectorBuilder<L::Service, TlsStage<PROXY>, D>
461 where
462 L: Layer<T>,
463 {
464 let connector = connector_layer.into_layer(self.connector);
465
466 EasyHttpConnectorBuilder {
467 dns: self.dns,
468 connector,
469 stage: Default::default(),
470 }
471 }
472
473 #[cfg(feature = "boring")]
474 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
475 pub fn with_tls_support_using_boringssl(
480 self,
481 config: TlsClientConfig,
482 ) -> EasyHttpConnectorBuilder<boring_client::TlsConnector<T>, TlsStage<PROXY>, D> {
483 let connector = boring_client::TlsConnector::auto(self.connector).with_base_config(config);
484
485 EasyHttpConnectorBuilder {
486 dns: self.dns,
487 connector,
488 stage: Default::default(),
489 }
490 }
491
492 #[cfg(feature = "boring")]
493 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
494 pub fn with_tls_support_using_boringssl_and_default_http_version(
504 self,
505 config: TlsClientConfig,
506 default_http_version: rama_http::Version,
507 ) -> EasyHttpConnectorBuilder<
508 AddInputExtension<boring_client::TlsConnector<T>, FallbackHttpVersion>,
509 TlsStage<PROXY>,
510 D,
511 > {
512 let connector = boring_client::TlsConnector::auto(self.connector).with_base_config(config);
513 let connector =
514 AddInputExtension::new(connector, FallbackHttpVersion(default_http_version))
515 .with_overwrite(false);
516
517 EasyHttpConnectorBuilder {
518 dns: self.dns,
519 connector,
520 stage: Default::default(),
521 }
522 }
523
524 #[cfg(feature = "rustls")]
525 #[cfg_attr(docsrs, doc(cfg(feature = "rustls")))]
526 pub fn with_tls_support_using_rustls(
531 self,
532 config: TlsClientConfig,
533 ) -> EasyHttpConnectorBuilder<rustls_client::TlsConnector<T>, TlsStage<PROXY>, D> {
534 let connector = rustls_client::TlsConnector::auto(self.connector).with_base_config(config);
535
536 EasyHttpConnectorBuilder {
537 dns: self.dns,
538 connector,
539 stage: Default::default(),
540 }
541 }
542
543 #[cfg(feature = "rustls")]
544 #[cfg_attr(docsrs, doc(cfg(feature = "rustls")))]
545 pub fn with_tls_support_using_rustls_and_default_http_version(
555 self,
556 config: TlsClientConfig,
557 default_http_version: rama_http::Version,
558 ) -> EasyHttpConnectorBuilder<
559 AddInputExtension<rustls_client::TlsConnector<T>, FallbackHttpVersion>,
560 TlsStage<PROXY>,
561 D,
562 > {
563 let connector = rustls_client::TlsConnector::auto(self.connector).with_base_config(config);
564 let connector =
565 AddInputExtension::new(connector, FallbackHttpVersion(default_http_version))
566 .with_overwrite(false);
567
568 EasyHttpConnectorBuilder {
569 dns: self.dns,
570 connector,
571 stage: Default::default(),
572 }
573 }
574
575 pub fn without_tls_support(self) -> EasyHttpConnectorBuilder<T, TlsStage<PROXY>, D> {
577 EasyHttpConnectorBuilder {
578 dns: self.dns,
579 connector: self.connector,
580 stage: Default::default(),
581 }
582 }
583}
584
585#[cfg(feature = "boring")]
586impl<T, D, const PROXY: bool>
587 EasyHttpConnectorBuilder<boring_client::TlsConnector<T>, TlsStage<PROXY>, D>
588{
589 #[cfg_attr(docsrs, doc(cfg(feature = "boring")))]
590 #[must_use]
596 pub fn with_tls_session_store(
597 mut self,
598 store: Arc<dyn boring_client::TlsClientSessionStore>,
599 ) -> Self {
600 self.connector.set_session_store(store);
601 self
602 }
603}
604
605impl<T, D, const PROXY: bool> EasyHttpConnectorBuilder<T, TlsStage<PROXY>, D> {
606 pub fn with_default_http_connector<Body>(
608 self,
609 scope: impl Into<WeakScope>,
610 ) -> EasyHttpConnectorBuilder<HttpConnector<T, Body>, HttpStage<PROXY>, D> {
611 let scope = scope.into();
612 let connector = HttpConnector::new(self.connector, scope);
613
614 EasyHttpConnectorBuilder {
615 dns: self.dns,
616 connector,
617 stage: HttpStage,
618 }
619 }
620
621 pub fn with_custom_http_connector<L>(
623 self,
624 connector_layer: L,
625 ) -> EasyHttpConnectorBuilder<L::Service, HttpStage<PROXY>, D>
626 where
627 L: Layer<T>,
628 {
629 let connector = connector_layer.into_layer(self.connector);
630
631 EasyHttpConnectorBuilder {
632 dns: self.dns,
633 connector,
634 stage: HttpStage,
635 }
636 }
637}
638
639#[cfg(feature = "quic")]
640#[cfg_attr(docsrs, doc(cfg(feature = "quic")))]
641impl<T, Body, D, const PROXY: bool>
642 EasyHttpConnectorBuilder<HttpConnector<T, Body>, HttpStage<PROXY>, D>
643{
644 pub fn with_http3_connector<C>(
650 self,
651 connector: C,
652 ) -> EasyHttpConnectorBuilder<
653 HttpConnector<super::HttpTransportConnector<T, C>, Body>,
654 HttpStage<PROXY>,
655 D,
656 > {
657 EasyHttpConnectorBuilder {
658 dns: self.dns,
659 connector: self.connector.with_http3_connector(connector),
660 stage: self.stage,
661 }
662 }
663
664 pub fn with_http3_support<E>(
670 self,
671 connector: super::Http3Connector<E>,
672 ) -> EasyHttpConnectorBuilder<
673 HttpConnector<super::HttpTransportConnector<T, D::Service>, Body>,
674 HttpStage<PROXY>,
675 D,
676 >
677 where
678 D: Layer<super::Http3Connector<E>>,
679 {
680 let connector = self.dns.layer(connector);
681 self.with_http3_connector(connector)
682 }
683
684 #[cfg(feature = "masque")]
685 #[cfg_attr(docsrs, doc(cfg(feature = "masque")))]
686 pub fn with_http3_masque_support<S>(
697 self,
698 connector: super::Http3ConnectorBuilder,
699 proxy_client: S,
700 ) -> Result<
701 EasyHttpConnectorBuilder<
702 HttpConnector<super::HttpTransportConnector<T, D::Service>, Body>,
703 HttpStage<PROXY>,
704 D,
705 >,
706 BoxError,
707 >
708 where
709 S: Clone,
710 D: Layer<super::Http3Connector<MasqueQuicEndpoints<S>>>,
711 {
712 let scope = connector.scope().clone();
713 let tunnels = ConnectUdpConnector::new(UdpConnector::new(), proxy_client, scope.clone());
714 let connector = connector
715 .with_endpoints(QuicEndpointConnector::new(tunnels, scope))
716 .build_lazy()?;
717 Ok(self.with_http3_support(connector))
718 }
719}
720
721#[cfg(all(feature = "masque", feature = "quic"))]
722#[cfg_attr(docsrs, doc(cfg(all(feature = "masque", feature = "quic"))))]
723pub type MasqueQuicEndpoints<S> = QuicEndpointConnector<ConnectUdpConnector<S, UdpConnector>>;
726
727type DefaultHttpConnector<T> = RequestVersionAdapter<
728 HttpConnectRequestAdapter<
729 MapEstablishedConnection<
730 HttpServiceConnector<ProxyRoutesConnector<T>>,
731 AltSvcConnectionLayer,
732 >,
733 >,
734>;
735
736#[doc(hidden)]
741#[derive(Clone, Debug)]
742pub struct AltSvcConnectionLayer {
743 cache: Option<AltSvcCache>,
744}
745
746impl<S> Layer<S> for AltSvcConnectionLayer {
747 type Service = AltSvc<S>;
748
749 fn layer(&self, inner: S) -> Self::Service {
750 match &self.cache {
751 Some(cache) => AltSvcLayer::new(cache.clone()).layer(inner),
752 None => AltSvc::passthrough(inner),
753 }
754 }
755}
756
757type ConfiguredConnectionBuilder<T> = EasyHttpConnectorBuilder<DefaultHttpConnector<T>, PoolStage>;
758
759type ConfiguredConnectionPoolBuilder<T> =
760 EasyHttpConnectorBuilder<DefaultHttpConnector<HttpPooledConnector<T>>, PoolStage>;
761
762type ErasedConnector<C> =
763 BoxService<ConnectRequest, EstablishedClientConnection<C, ConnectRequest>, ConnectionError>;
764
765type DefaultConnectionBuilder<C> =
766 ConfiguredConnectionBuilder<ProxyRouteFailureCacheConnector<ErasedConnector<C>>>;
767
768type DefaultConnectionPoolBuilder<C> =
769 ConfiguredConnectionPoolBuilder<ProxyRouteFailureCacheConnector<ErasedConnector<C>>>;
770
771#[derive(ExtensionsRef)]
776struct ConnectorServiceAdapter<T>(T);
777
778impl<T> Service<ConnectRequest> for ConnectorServiceAdapter<T>
779where
780 T: ConnectorService<ConnectRequest>,
781{
782 type Output = EstablishedClientConnection<T::Connection, ConnectRequest>;
783 type Error = ConnectionError;
784
785 fn serve(
786 &self,
787 input: ConnectRequest,
788 ) -> impl Future<Output = Result<Self::Output, Self::Error>> + Send + '_ {
789 self.0.connect(input)
790 }
791}
792
793fn erase_connector<T>(connector: T) -> ErasedConnector<T::Connection>
794where
795 T: ConnectorService<ConnectRequest>,
796{
797 ConnectorServiceAdapter(connector).boxed()
798}
799
800fn finalize_http_connector<T>(connector: T) -> DefaultHttpConnector<T> {
801 let connector = ProxyRoutesConnector::new(connector);
802 let cache = AltSvcCache::default();
803 let connector = HttpServiceConnector::new(connector).with_cache(cache.clone());
804 adapt_http_service_connector(connector, Some(cache))
805}
806
807fn adapt_http_service_connector<T>(
808 connector: HttpServiceConnector<ProxyRoutesConnector<T>>,
809 cache: Option<AltSvcCache>,
810) -> DefaultHttpConnector<T> {
811 let connector = MapEstablishedConnection::new(connector, AltSvcConnectionLayer { cache });
812 RequestVersionAdapter::new(HttpConnectRequestAdapter::new(connector))
813}
814
815impl<T> EasyHttpConnectorBuilder<DefaultHttpConnector<T>, PoolStage> {
816 #[must_use]
818 pub fn without_alt_svc(self) -> Self {
819 self.maybe_with_alt_svc_cache(None)
820 }
821
822 generate_set_and_with! {
823 pub fn alt_svc_cache(mut self, cache: Option<AltSvcCache>) -> Self {
825 let connector = self.connector.get_mut().get_mut();
826 connector.get_mut().maybe_set_cache(cache.clone());
827 connector.layer_mut().cache = cache;
828 self
829 }
830 }
831}
832
833fn finish_without_connection_pool<T, Stage, D>(
834 builder: EasyHttpConnectorBuilder<T, Stage, D>,
835) -> ConfiguredConnectionBuilder<T>
836where
837 T: ConnectorService<ConnectRequest>,
838{
839 EasyHttpConnectorBuilder {
840 dns: (),
841 connector: finalize_http_connector(builder.connector),
842 stage: Default::default(),
843 }
844}
845
846fn finish_with_connection_pool<T, Stage, D>(
847 builder: EasyHttpConnectorBuilder<T, Stage, D>,
848 config: HttpPooledConnectorConfig,
849) -> ConfiguredConnectionPoolBuilder<T>
850where
851 T: ConnectorService<ConnectRequest>,
852{
853 EasyHttpConnectorBuilder {
854 dns: (),
855 connector: finalize_http_connector(config.build_connector(builder.connector)),
856 stage: Default::default(),
857 }
858}
859
860fn finish_with_custom_connection_pool<T, Stage, D, P, R>(
861 builder: EasyHttpConnectorBuilder<T, Stage, D>,
862 pool: P,
863 req_to_conn_id: R,
864 wait_for_pool_timeout: Option<Duration>,
865) -> EasyHttpConnectorBuilder<PooledConnector<T, P, R>, PoolStage> {
866 let connector = PooledConnector::new(builder.connector, pool, req_to_conn_id)
867 .maybe_with_wait_for_pool_timeout(wait_for_pool_timeout);
868 EasyHttpConnectorBuilder {
869 dns: (),
870 connector,
871 stage: Default::default(),
872 }
873}
874
875impl<T, D, const PROXY: bool> EasyHttpConnectorBuilder<T, HttpStage<PROXY>, D> {
876 #[must_use]
882 pub fn with_proxy_route_failure_cache(
883 self,
884 cache: ProxyRouteFailureCache,
885 ) -> EasyHttpConnectorBuilder<
886 ProxyRouteFailureCacheConnector<ErasedConnector<T::Connection>>,
887 ProxyRouteFailureCacheStage,
888 D,
889 >
890 where
891 T: ConnectorService<ConnectRequest>,
892 {
893 EasyHttpConnectorBuilder {
894 dns: self.dns,
895 connector: ProxyRouteFailureCacheConnector::new(erase_connector(self.connector), cache),
896 stage: ProxyRouteFailureCacheStage,
897 }
898 }
899
900 #[must_use]
902 pub fn without_proxy_route_failure_cache(
903 self,
904 ) -> EasyHttpConnectorBuilder<T, ProxyRouteFailureCacheStage, D> {
905 EasyHttpConnectorBuilder {
906 dns: self.dns,
907 connector: self.connector,
908 stage: ProxyRouteFailureCacheStage,
909 }
910 }
911}
912
913impl<T, D> EasyHttpConnectorBuilder<T, HttpStage<true>, D> {
914 pub fn without_connection_pool(self) -> DefaultConnectionBuilder<T::Connection>
920 where
921 T: ConnectorService<ConnectRequest>,
922 {
923 finish_without_connection_pool(
924 self.with_proxy_route_failure_cache(ProxyRouteFailureCache::default()),
925 )
926 }
927
928 pub fn with_connection_pool(
949 self,
950 config: HttpPooledConnectorConfig,
951 ) -> DefaultConnectionPoolBuilder<T::Connection>
952 where
953 T: ConnectorService<ConnectRequest>,
954 {
955 finish_with_connection_pool(
956 self.with_proxy_route_failure_cache(ProxyRouteFailureCache::default()),
957 config,
958 )
959 }
960
961 pub fn with_default_connection_pool(self) -> DefaultConnectionPoolBuilder<T::Connection>
964 where
965 T: ConnectorService<ConnectRequest>,
966 {
967 self.with_connection_pool(HttpPooledConnectorConfig::default())
968 }
969
970 pub fn with_custom_connection_pool<P, R>(
996 self,
997 pool: P,
998 req_to_conn_id: R,
999 wait_for_pool_timeout: Option<Duration>,
1000 ) -> EasyHttpConnectorBuilder<
1001 PooledConnector<ProxyRouteFailureCacheConnector<ErasedConnector<T::Connection>>, P, R>,
1002 PoolStage,
1003 >
1004 where
1005 T: ConnectorService<ConnectRequest>,
1006 {
1007 finish_with_custom_connection_pool(
1008 self.with_proxy_route_failure_cache(ProxyRouteFailureCache::default()),
1009 pool,
1010 req_to_conn_id,
1011 wait_for_pool_timeout,
1012 )
1013 }
1014}
1015
1016impl<T, D> EasyHttpConnectorBuilder<T, HttpStage<false>, D> {
1017 pub fn without_connection_pool(self) -> ConfiguredConnectionBuilder<T>
1023 where
1024 T: ConnectorService<ConnectRequest>,
1025 {
1026 finish_without_connection_pool(self)
1027 }
1028
1029 pub fn with_connection_pool(
1031 self,
1032 config: HttpPooledConnectorConfig,
1033 ) -> ConfiguredConnectionPoolBuilder<T>
1034 where
1035 T: ConnectorService<ConnectRequest>,
1036 {
1037 finish_with_connection_pool(self, config)
1038 }
1039
1040 pub fn with_default_connection_pool(self) -> ConfiguredConnectionPoolBuilder<T>
1043 where
1044 T: ConnectorService<ConnectRequest>,
1045 {
1046 finish_with_connection_pool(self, HttpPooledConnectorConfig::default())
1047 }
1048
1049 pub fn with_custom_connection_pool<P, R>(
1056 self,
1057 pool: P,
1058 req_to_conn_id: R,
1059 wait_for_pool_timeout: Option<Duration>,
1060 ) -> EasyHttpConnectorBuilder<PooledConnector<T, P, R>, PoolStage> {
1061 finish_with_custom_connection_pool(self, pool, req_to_conn_id, wait_for_pool_timeout)
1062 }
1063}
1064
1065impl<T, D> EasyHttpConnectorBuilder<T, ProxyRouteFailureCacheStage, D> {
1066 pub fn without_connection_pool(self) -> ConfiguredConnectionBuilder<T>
1068 where
1069 T: ConnectorService<ConnectRequest>,
1070 {
1071 finish_without_connection_pool(self)
1072 }
1073
1074 pub fn with_connection_pool(
1076 self,
1077 config: HttpPooledConnectorConfig,
1078 ) -> ConfiguredConnectionPoolBuilder<T>
1079 where
1080 T: ConnectorService<ConnectRequest>,
1081 {
1082 finish_with_connection_pool(self, config)
1083 }
1084
1085 pub fn with_default_connection_pool(self) -> ConfiguredConnectionPoolBuilder<T>
1088 where
1089 T: ConnectorService<ConnectRequest>,
1090 {
1091 finish_with_connection_pool(self, HttpPooledConnectorConfig::default())
1092 }
1093
1094 pub fn with_custom_connection_pool<P, R>(
1107 self,
1108 pool: P,
1109 req_to_conn_id: R,
1110 wait_for_pool_timeout: Option<Duration>,
1111 ) -> EasyHttpConnectorBuilder<PooledConnector<T, P, R>, PoolStage> {
1112 finish_with_custom_connection_pool(self, pool, req_to_conn_id, wait_for_pool_timeout)
1113 }
1114}
1115
1116impl<T> EasyHttpConnectorBuilder<T, PoolStage> {
1117 pub fn build_client_with_early_data<Body, ModifiedBody, ConnResponse, IncomingBody, R, P>(
1142 self,
1143 application_rule: R,
1144 prepare_body: P,
1145 ) -> impl Service<Request<Body>, Output = Response, Error = RetryError>
1146 where
1147 Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Send + 'static,
1148 ModifiedBody:
1149 StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
1150 T: Service<
1151 Request<RetryBody>,
1152 Output = EstablishedClientConnection<ConnResponse, Request<ModifiedBody>>,
1153 Error: Into<BoxError>,
1154 >,
1155 ConnResponse: Service<Request<ModifiedBody>, Output = Response<IncomingBody>, Error = BoxError>
1156 + ExtensionsRef,
1157 IncomingBody: Send + Sync + 'static,
1158 ResponseBody: From<IncomingBody>,
1159 R: RetryRule<Request<RetryBody>, Response, OpaqueError>,
1160 P: PrepareBody,
1161 {
1162 let client = self.build_client::<RetryBody, ModifiedBody, ConnResponse>();
1163 let rule = with_recovery(application_rule, |error: &OpaqueError| {
1164 is_early_data_rejected(error)
1165 });
1166 Retry::new(
1167 ManagedPolicy::new(rule).with_prepare_body(prepare_body),
1168 MapInput::new(client, permit_attempt),
1169 )
1170 }
1171
1172 pub fn build_client<Body, ModifiedBody, ConnResponse>(
1174 self,
1175 ) -> super::EasyHttpWebClient<Body, T::Output, ()>
1176 where
1177 Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
1178 ModifiedBody:
1179 StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
1180 T: Service<
1181 Request<Body>,
1182 Output = EstablishedClientConnection<ConnResponse, Request<ModifiedBody>>,
1183 Error: Into<BoxError>,
1184 >,
1185 ConnResponse: ExtensionsRef,
1186 {
1187 super::EasyHttpWebClient::new(self.connector)
1188 }
1189}
1190
1191impl<T, S, D> EasyHttpConnectorBuilder<T, S, D> {
1192 pub fn build_connector(self) -> T {
1194 self.connector
1195 }
1196}
1197
1198#[cfg(test)]
1199mod tests {
1200 use super::*;
1201 use crate::http::Body;
1202 use rama_utils::octets::kib;
1203
1204 use rama_core::layer::layer_fn;
1205 #[cfg(all(
1206 feature = "quic",
1207 any(
1208 feature = "boring",
1209 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1210 )
1211 ))]
1212 use {
1213 crate::http::client::Http3Connector,
1214 crate::quic::Endpoint,
1215 rama_core::futures::{Stream, stream},
1216 rama_http::layer::http_service::HttpServiceAttempt,
1217 rama_net::{
1218 address::{Domain, SocketAddress},
1219 client::ConnectionErrorKind,
1220 http::TargetHttpVersion as RequestedVersion,
1221 },
1222 std::{
1223 net::{Ipv4Addr, Ipv6Addr, SocketAddr},
1224 sync::{
1225 Arc,
1226 atomic::{AtomicUsize, Ordering},
1227 },
1228 },
1229 };
1230 #[cfg(feature = "tls")]
1231 use {
1232 crate::http::client::HttpConnIdentifier,
1233 crate::http::{
1234 HeaderMap, Version,
1235 conn::{HttpOrigin, SelectedHttpService},
1236 header::ALT_SVC,
1237 },
1238 rama_core::{ServiceInput, service::service_fn},
1239 rama_net::{
1240 Protocol,
1241 address::HostWithPort,
1242 client::{
1243 ConnectRequest,
1244 pool::{ConnID as _, ReqToConnID},
1245 },
1246 },
1247 rama_net::{http::TargetHttpVersion, tls::ApplicationProtocol},
1248 rama_tls::client::{ServerVerifyMode, TlsPoolId, TlsServerVerify},
1249 rama_tls::{
1250 ProtocolVersion,
1251 client::{NegotiatedTlsParameters, TlsServerAuthentication},
1252 },
1253 std::convert::Infallible,
1254 };
1255
1256 fn assert_future_budget(connector: &impl Service<Request>, name: &str) {
1257 let request = Request::builder()
1258 .uri("https://example.com/")
1259 .body(Body::empty())
1260 .unwrap();
1261 let future = connector.serve(request);
1262 let size = std::mem::size_of_val(&future);
1263 assert!(
1264 size <= kib(64),
1265 "{name} connector future is {size} bytes; connector error adapters must not duplicate nested future storage"
1266 );
1267 }
1268
1269 #[tokio::test]
1270 async fn default_connector_futures_stay_within_stack_budget() {
1271 let builder = || {
1272 EasyHttpConnectorBuilder::new()
1273 .with_default_transport_connector()
1274 .without_dns_connector()
1275 .without_tls_proxy_support()
1276 .without_proxy_support()
1277 };
1278 let plain = builder()
1279 .without_tls_support()
1280 .with_default_http_connector::<Body>(WeakScope::detached())
1281 .with_default_connection_pool()
1282 .build_connector();
1283 assert_future_budget(&plain, "plain pooled");
1284 #[cfg(feature = "rustls")]
1285 {
1286 let pooled = builder()
1287 .with_tls_support_using_rustls(TlsClientConfig::new())
1288 .with_default_http_connector::<Body>(WeakScope::detached())
1289 .with_default_connection_pool()
1290 .build_connector();
1291 assert_future_budget(&pooled, "rustls pooled");
1292 let unpooled = builder()
1293 .with_tls_support_using_rustls(TlsClientConfig::new())
1294 .with_default_http_connector::<Body>(WeakScope::detached())
1295 .without_connection_pool()
1296 .build_connector();
1297 assert_future_budget(&unpooled, "rustls unpooled");
1298 }
1299 }
1300
1301 #[cfg(feature = "tls")]
1302 #[tokio::test]
1303 async fn custom_http_connector_discovers_h3_without_capability_configuration() {
1304 let cache = AltSvcCache::default();
1305 let origin =
1306 HttpOrigin::new(Protocol::HTTPS, HostWithPort::example_domain_https()).unwrap();
1307 let mut headers = HeaderMap::new();
1308 headers.insert(ALT_SVC, "h3=\":8443\"".parse().unwrap());
1309 cache.record(&origin, &headers, Duration::ZERO);
1310
1311 let builder = EasyHttpConnectorBuilder::new()
1312 .with_custom_transport_connector(())
1313 .without_dns_connector()
1314 .without_tls_proxy_support()
1315 .without_proxy_support()
1316 .without_tls_support()
1317 .with_custom_http_connector(layer_fn(|()| {
1318 service_fn(|input: ConnectRequest| async move {
1319 let version = input
1320 .extensions()
1321 .get_ref::<TargetHttpVersion>()
1322 .map_or(Version::HTTP_2, |version| version.0);
1323 let conn = ServiceInput::new(());
1324 conn.extensions.insert(TargetHttpVersion(version));
1325 conn.extensions
1326 .insert(TlsServerAuthentication(Some(input.authority.host.clone())));
1327 conn.extensions.insert(NegotiatedTlsParameters {
1328 protocol_version: ProtocolVersion::TLSv1_3,
1329 application_layer_protocol: Some(if version == Version::HTTP_3 {
1330 ApplicationProtocol::HTTP_3
1331 } else {
1332 ApplicationProtocol::HTTP_2
1333 }),
1334 peer_certificate_chain: None,
1335 server_name: None,
1336 resumed: None,
1337 algorithms: Default::default(),
1338 server_hello: None,
1339 });
1340 Ok::<_, Infallible>(EstablishedClientConnection { input, conn })
1341 })
1342 }))
1343 .without_connection_pool()
1344 .with_alt_svc_cache(cache.clone());
1345 let connector = builder.build_connector();
1346 for _ in 0..2 {
1347 let established = connector
1348 .serve(
1349 Request::builder()
1350 .uri("https://example.com/")
1351 .body(Body::empty())
1352 .unwrap(),
1353 )
1354 .await
1355 .unwrap();
1356 assert!(
1357 established
1358 .input
1359 .extensions()
1360 .contains::<SelectedHttpService>()
1361 );
1362 assert_eq!(established.input.version(), Version::HTTP_3);
1363 assert_eq!(
1364 established
1365 .input
1366 .extensions()
1367 .get_ref::<SelectedHttpService>()
1368 .unwrap()
1369 .candidate
1370 .target
1371 .port,
1372 8443
1373 );
1374 }
1375 }
1376
1377 #[test]
1378 fn custom_tls_proxy_layer_requires_no_builtin_provider() {
1379 EasyHttpConnectorBuilder::new()
1380 .with_custom_transport_connector(())
1381 .without_dns_connector()
1382 .with_custom_tls_proxy_connector(layer_fn(|inner| inner))
1383 .build_connector();
1384 }
1385
1386 #[cfg(feature = "tls")]
1387 #[test]
1388 fn custom_fixed_policy_pool_respects_explicit_override_identity() {
1389 let identifier = HttpConnIdentifier::default();
1390 let input = ConnectRequest::new(HostWithPort::example_domain_https());
1391 let fixed = identifier.id(&input).unwrap();
1392 assert!(fixed.is_reusable());
1393 let policy = TlsPoolId::builder()
1394 .with_verify(&TlsServerVerify(ServerVerifyMode::Disable))
1395 .build()
1396 .unwrap();
1397 input.extensions.insert(policy);
1398 let overridden = identifier.id(&input).unwrap();
1399 assert!(overridden.is_reusable());
1400 assert_ne!(fixed, overridden);
1401 assert_eq!(overridden, identifier.id(&input).unwrap());
1402 input.extensions.insert(TlsPoolId::non_reusable());
1403 assert!(!identifier.id(&input).unwrap().is_reusable());
1404 }
1405
1406 #[cfg(all(
1407 feature = "quic",
1408 any(
1409 feature = "boring",
1410 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1411 )
1412 ))]
1413 #[derive(Clone)]
1414 struct RecordingResolver {
1415 calls: Arc<AtomicUsize>,
1416 loopback: bool,
1417 }
1418
1419 #[cfg(all(
1420 feature = "quic",
1421 any(
1422 feature = "boring",
1423 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1424 )
1425 ))]
1426 impl DnsAddressResolver for RecordingResolver {
1427 type Error = Infallible;
1428
1429 fn lookup_ipv4(
1430 &self,
1431 domain: Domain,
1432 ) -> impl Stream<Item = Result<Ipv4Addr, Self::Error>> + Send + '_ {
1433 assert_eq!(domain.as_str(), "private.invalid");
1434 self.calls.fetch_add(1, Ordering::Relaxed);
1435 stream::iter(self.loopback.then_some(Ok(Ipv4Addr::LOCALHOST)))
1436 }
1437
1438 fn lookup_ipv6(
1439 &self,
1440 domain: Domain,
1441 ) -> impl Stream<Item = Result<Ipv6Addr, Self::Error>> + Send + '_ {
1442 assert_eq!(domain.as_str(), "private.invalid");
1443 self.calls.fetch_add(1, Ordering::Relaxed);
1444 stream::iter(self.loopback.then_some(Ok(Ipv6Addr::LOCALHOST)))
1445 }
1446 }
1447
1448 #[cfg(all(
1449 feature = "quic",
1450 any(
1451 feature = "boring",
1452 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1453 )
1454 ))]
1455 struct RecordingDnsLayer {
1456 resolver: RecordingResolver,
1457 applied: Arc<AtomicUsize>,
1458 }
1459
1460 #[cfg(all(
1461 feature = "quic",
1462 any(
1463 feature = "boring",
1464 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1465 )
1466 ))]
1467 impl<S> Layer<S> for RecordingDnsLayer {
1468 type Service = DnsConnector<S, RecordingResolver>;
1469
1470 fn layer(&self, inner: S) -> Self::Service {
1471 self.applied.fetch_add(1, Ordering::Relaxed);
1472 DnsConnector::with_resolver(inner, self.resolver.clone())
1473 }
1474 }
1475
1476 #[cfg(all(
1477 feature = "quic",
1478 any(
1479 feature = "boring",
1480 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1481 )
1482 ))]
1483 #[tokio::test]
1484 async fn built_in_quic_uses_the_configured_dns_resolver() {
1485 let calls = Arc::new(AtomicUsize::new(0));
1486 let applied = Arc::new(AtomicUsize::new(0));
1487 let scope = WeakScope::detached();
1488 let h3 = Http3Connector::builder(scope.clone())
1489 .build()
1490 .await
1491 .unwrap();
1492 let connector = EasyHttpConnectorBuilder::new()
1493 .with_default_transport_connector()
1494 .with_dns_connector(RecordingDnsLayer {
1495 resolver: RecordingResolver {
1496 calls: calls.clone(),
1497 loopback: false,
1498 },
1499 applied: applied.clone(),
1500 })
1501 .without_tls_proxy_support()
1502 .without_proxy_support()
1503 .without_tls_support()
1504 .with_default_http_connector::<Body>(scope)
1505 .with_http3_support(h3)
1506 .build_connector();
1507 let input = ConnectRequest::new("private.invalid:443".parse().unwrap())
1508 .with_application_protocol(Protocol::HTTPS);
1509 input.extensions.insert(RequestedVersion(Version::HTTP_3));
1510 let error = connector.serve(input).await.err().unwrap();
1511 assert_eq!(error.kind(), ConnectionErrorKind::Unavailable);
1512 assert!(calls.load(Ordering::Relaxed) > 0);
1513 assert_eq!(
1514 applied.load(Ordering::Relaxed),
1515 2,
1516 "the same DNS layer must wrap both transports"
1517 );
1518 }
1519
1520 #[cfg(all(
1521 feature = "quic",
1522 any(
1523 feature = "boring",
1524 all(feature = "rustls", any(feature = "ring", feature = "aws-lc"))
1525 )
1526 ))]
1527 #[tokio::test]
1528 async fn ipv4_quic_endpoint_skips_ipv6_candidates_without_terminal_failure() {
1529 let scope = WeakScope::detached();
1530 let endpoint = Endpoint::bind_client(scope.clone(), SocketAddress::local_ipv4(0))
1531 .await
1532 .unwrap();
1533 let blackhole = tokio::net::UdpSocket::bind(SocketAddr::from(SocketAddress::local_ipv4(0)))
1534 .await
1535 .unwrap();
1536 let connector = Http3Connector::builder(scope)
1537 .with_endpoint(endpoint.clone())
1538 .build()
1539 .await
1540 .unwrap();
1541 let calls = Arc::new(AtomicUsize::new(0));
1542 let connector = DnsConnector::with_resolver(
1543 connector,
1544 RecordingResolver {
1545 calls: calls.clone(),
1546 loopback: true,
1547 },
1548 );
1549 let input = ConnectRequest::new(
1550 format!("private.invalid:{}", blackhole.local_addr().unwrap().port())
1551 .parse()
1552 .unwrap(),
1553 )
1554 .with_application_protocol(Protocol::HTTPS);
1555 let attempt = Arc::new(HttpServiceAttempt::default());
1556 input.extensions.insert_arc(attempt.clone());
1557 tokio::time::timeout(Duration::from_millis(500), connector.serve(input))
1558 .await
1559 .unwrap_err();
1560 assert!(calls.load(Ordering::Relaxed) >= 2);
1561 assert!(
1562 !attempt.failed(),
1563 "a mismatched candidate family must not turn the IPv4 timeout into an authentication failure"
1564 );
1565 endpoint.close(0u32, b"test complete");
1566 }
1567}