rama/http/client/
mod.rs

1//! rama http client support
2//!
3//! Contains re-exports from `rama-http-backend::client`
4//! and adds `EasyHttpWebClient`, an opiniated http web client which
5//! supports most common use cases and provides sensible defaults.
6use std::{fmt, io};
7
8use crate::{
9    Layer, Service,
10    error::BoxError,
11    extensions::ExtensionsRef,
12    http::{Request, Response, StreamingBody},
13    net::client::EstablishedClientConnection,
14    rt::Executor,
15    service::BoxService,
16    telemetry::tracing,
17};
18
19#[doc(inline)]
20pub use ::rama_http::service::client::blocking::{
21    Body as BlockingBody, Client as BlockingHttpClient, Response as BlockingResponse,
22};
23#[doc(inline)]
24pub use ::rama_http_backend::client::*;
25use rama_core::{
26    error::{ErrorContext, ErrorExt as _, extra::OpaqueError},
27    extensions::Egress,
28    layer::MapErr,
29};
30
31pub mod builder;
32#[doc(inline)]
33pub use builder::EasyHttpConnectorBuilder;
34
35#[cfg(feature = "socks5")]
36mod proxy_connector;
37#[cfg(feature = "socks5")]
38#[cfg_attr(docsrs, doc(cfg(feature = "socks5")))]
39#[doc(inline)]
40pub use proxy_connector::{MaybeProxiedConnection, ProxyConnector, ProxyConnectorLayer};
41
42/// An opiniated http client that can be used to serve HTTP requests.
43///
44/// Use [`EasyHttpWebClient::connector_builder()`] to easily create a client with
45/// a common Http connector setup (tcp + proxy + tls + http) or bring your
46/// own http connector.
47///
48/// [`Default`] uses Rama's default multiplexing connection pool. Build the
49/// connector explicitly with
50/// [`EasyHttpConnectorBuilder::without_connection_pool`] when connection reuse
51/// is unwanted.
52///
53/// You can fork this http client in case you have use cases not possible with this service example.
54/// E.g. perhaps you wish to have middleware in into outbound requests, after they
55/// passed through your "connector" setup. All this and more is possible by defining your own
56/// http client. Rama is here to empower you, the building blocks are there, go crazy
57/// with your own service fork and use the full power of Rust at your fingertips ;)
58pub struct EasyHttpWebClient<BodyIn, ConnResponse, L> {
59    connector: BoxService<Request<BodyIn>, ConnResponse, OpaqueError>,
60    jit_layers: L,
61}
62
63impl<BodyIn, ConnResponse, L> fmt::Debug for EasyHttpWebClient<BodyIn, ConnResponse, L> {
64    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
65        f.debug_struct("EasyHttpWebClient").finish()
66    }
67}
68
69impl<BodyIn, ConnResponse, L: Clone> Clone for EasyHttpWebClient<BodyIn, ConnResponse, L> {
70    fn clone(&self) -> Self {
71        Self {
72            connector: self.connector.clone(),
73            jit_layers: self.jit_layers.clone(),
74        }
75    }
76}
77
78impl EasyHttpWebClient<(), (), ()> {
79    /// Create a [`EasyHttpConnectorBuilder`] to easily create a [`EasyHttpWebClient`] with a custom connector
80    #[must_use]
81    pub fn connector_builder() -> EasyHttpConnectorBuilder {
82        EasyHttpConnectorBuilder::new()
83    }
84
85    /// Create a cloneable blocking HTTP(S) client with its own dedicated
86    /// runtime thread and Rama's default web connector stack.
87    ///
88    /// ```no_run
89    /// use rama::http::client::EasyHttpWebClient;
90    ///
91    /// # fn main() -> Result<(), rama::error::BoxError> {
92    /// let client = EasyHttpWebClient::try_blocking()?;
93    /// let client_for_worker = client.clone();
94    ///
95    /// let text = client_for_worker
96    ///     .get("https://example.com/")
97    ///     .send()?
98    ///     .try_into_string()?;
99    /// # _ = text;
100    /// # Ok(())
101    /// # }
102    /// ```
103    pub fn try_blocking() -> io::Result<BlockingHttpWebClient> {
104        BlockingHttpClient::try_new(EasyHttpWebClient::default())
105    }
106}
107
108/// Rama's default asynchronous HTTP(S) client, including its default
109/// multiplexing connection pool.
110pub type DefaultHttpWebClient<Body = crate::http::Body> = EasyHttpWebClient<
111    Body,
112    EstablishedClientConnection<
113        BindBodyToConn<
114            crate::net::client::pool::MultiplexedConnection<
115                HttpClientService<Body>,
116                BasicHttpConId,
117            >,
118        >,
119        Request<Body>,
120    >,
121    (),
122>;
123
124/// A blocking HTTP(S) client using Rama's default pooled web connector stack.
125pub type BlockingHttpWebClient = BlockingHttpClient<DefaultHttpWebClient>;
126
127impl<Body> Default for DefaultHttpWebClient<Body>
128where
129    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
130{
131    #[inline(always)]
132    fn default() -> Self {
133        Self::default_with_executor(Executor::default())
134    }
135}
136
137impl<Body> DefaultHttpWebClient<Body>
138where
139    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
140{
141    core::cfg_select! {
142        feature = "boring" => {
143            pub fn default_with_executor(exec: Executor) -> Self {
144                let tls_config = crate::tls::client::TlsClientConfig::default_http();
145
146                EasyHttpConnectorBuilder::new()
147                    .with_default_transport_connector()
148                    .with_default_dns_connector()
149                    .with_tls_proxy_support_using_boringssl()
150                    .with_proxy_support()
151                    .with_tls_support_using_boringssl(tls_config)
152                    .with_default_http_connector(exec)
153                    .with_default_connection_pool()
154                    .build_client()
155            }
156        }
157        feature = "rustls" => {
158            pub fn default_with_executor(exec: Executor) -> Self {
159                let tls_config = crate::tls::client::TlsClientConfig::default_http();
160
161                EasyHttpConnectorBuilder::new()
162                    .with_default_transport_connector()
163                    .with_default_dns_connector()
164                    .with_tls_proxy_support_using_rustls()
165                    .with_proxy_support()
166                    .with_tls_support_using_rustls(tls_config)
167                    .with_default_http_connector(exec)
168                    .with_default_connection_pool()
169                    .build_client()
170            }
171        }
172        _ => {
173            pub fn default_with_executor(exec: Executor) -> Self {
174                EasyHttpConnectorBuilder::new()
175                    .with_default_transport_connector()
176                    .with_default_dns_connector()
177                    .without_tls_proxy_support()
178                    .with_proxy_support()
179                    .without_tls_support()
180                    .with_default_http_connector(exec)
181                    .with_default_connection_pool()
182                    .build_client()
183            }
184        }
185    }
186}
187
188impl<BodyIn, ConnResponse> EasyHttpWebClient<BodyIn, ConnResponse, ()>
189where
190    BodyIn: Send + 'static,
191{
192    /// Create a new [`EasyHttpWebClient`] using the provided connector
193    #[must_use]
194    pub fn new<S>(connector: S) -> Self
195    where
196        S: Service<Request<BodyIn>, Output = ConnResponse, Error: Into<BoxError>>,
197    {
198        Self {
199            connector: MapErr::into_opaque_error(connector).boxed(),
200            jit_layers: (),
201        }
202    }
203}
204
205impl<BodyIn, ConnResponse, L> EasyHttpWebClient<BodyIn, ConnResponse, L> {
206    /// Convert this asynchronous web client into a cloneable blocking client
207    /// with its own dedicated runtime thread.
208    pub fn try_into_blocking(self) -> io::Result<BlockingHttpClient<Self>> {
209        BlockingHttpClient::try_new(self)
210    }
211
212    /// Convert this asynchronous web client into a blocking client using a
213    /// caller-supplied runtime.
214    #[must_use]
215    pub fn into_blocking_with_runtime(
216        self,
217        runtime: &crate::rt::blocking::Runtime,
218    ) -> BlockingHttpClient<Self> {
219        BlockingHttpClient::with_runtime(self, runtime)
220    }
221
222    /// Set the connector that this [`EasyHttpWebClient`] will use
223    #[must_use]
224    pub fn with_connector<S, BodyInNew, ConnResponseNew>(
225        self,
226        connector: S,
227    ) -> EasyHttpWebClient<BodyInNew, ConnResponseNew, L>
228    where
229        S: Service<Request<BodyInNew>, Output = ConnResponseNew, Error: Into<BoxError>>,
230        BodyInNew: Send + 'static,
231    {
232        EasyHttpWebClient {
233            connector: MapErr::into_opaque_error(connector).boxed(),
234            jit_layers: self.jit_layers,
235        }
236    }
237
238    /// [`Layer`] which will be applied just in time (JIT) before the request is send, but after
239    /// the connection has been established.
240    ///
241    /// Simplified flow of how the [`EasyHttpWebClient`] works:
242    /// 1. External: let response = client.serve(request)
243    /// 2. Internal: let http_connection = self.connector.serve(request)
244    /// 3. Internal: let response = jit_layers.layer(http_connection).serve(request)
245    pub fn with_jit_layer<T>(self, jit_layers: T) -> EasyHttpWebClient<BodyIn, ConnResponse, T> {
246        EasyHttpWebClient {
247            connector: self.connector,
248            jit_layers,
249        }
250    }
251}
252
253impl<Body, ConnectionBody, Connection, L> Service<Request<Body>>
254    for EasyHttpWebClient<Body, EstablishedClientConnection<Connection, Request<ConnectionBody>>, L>
255where
256    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
257    Connection:
258        Service<Request<ConnectionBody>, Output = Response, Error = BoxError> + ExtensionsRef,
259    // Body type this connection will be able to send, this is not necessarily the same one that
260    // was used in the request that created this connection
261    ConnectionBody:
262        StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
263    L: Layer<
264            Connection,
265            Service: Service<Request<ConnectionBody>, Output = Response, Error = BoxError>,
266        > + Send
267        + Sync
268        + 'static,
269{
270    type Output = Response;
271    type Error = OpaqueError;
272
273    async fn serve(&self, req: Request<Body>) -> Result<Self::Output, Self::Error> {
274        let uri = req.uri().clone();
275
276        let EstablishedClientConnection {
277            input: req,
278            conn: http_connection,
279        } = self.connector.serve(req).await.into_opaque_error()?;
280
281        req.extensions()
282            .insert(Egress(http_connection.extensions().clone()));
283
284        let http_connection = self.jit_layers.layer(http_connection);
285
286        // NOTE: stack might change request version based on connector data,
287        tracing::trace!(url.full = %uri, "send http req to connector stack");
288
289        let result = http_connection.serve(req).await;
290
291        match result {
292            Ok(resp) => {
293                tracing::trace!(url.full = %uri, "response received from connector stack");
294                Ok(resp)
295            }
296            Err(err) => Err(err
297                .context("http request failure")
298                .context_field("uri", uri)
299                .into_opaque_error()),
300        }
301    }
302}
303
304#[cfg(test)]
305mod tests {
306    use std::{
307        convert::Infallible,
308        sync::{
309            Arc,
310            atomic::{AtomicUsize, Ordering},
311        },
312        time::Duration,
313    };
314
315    use rama_core::{error::BoxErrorExt as _, service::service_fn};
316    use rama_http::{Body, BodyExtractExt, Version};
317    use rama_http_backend::server::HttpServer;
318    use rama_net::{
319        address::ProxyAddress,
320        client::{
321            ConnectRequest, ConnectionError, ConnectionErrorKind, ConnectorService, ProxyRoute,
322            ProxyRouteFailureCache, ProxyRouteFailureCacheConfig, ProxyRouteFailureCacheScope,
323            ProxyRoutes,
324        },
325        test_utils::client::{MockConnectorService, MockSocket},
326    };
327    use serde::{Deserialize, Serialize};
328    use tokio::time::sleep;
329
330    use super::*;
331
332    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
333    struct Output {
334        conn: usize,
335        resp: usize,
336    }
337
338    fn dummy_server<Input: Send + 'static>()
339    -> impl Service<
340        Input,
341        Output = EstablishedClientConnection<MockSocket, Input>,
342        Error = Infallible,
343    > + Clone {
344        let created_connections = Arc::new(AtomicUsize::new(0));
345        MockConnectorService::new(move || {
346            let created_connections = created_connections.clone();
347            let conn = created_connections.fetch_add(1, Ordering::Relaxed);
348
349            // count responses created on this specific connection
350            let created_response = Arc::new(AtomicUsize::new(0));
351
352            HttpServer::auto(Executor::default()).service(service_fn(move |_req: Request| {
353                let created_response = created_response.clone();
354                let resp = created_response.fetch_add(1, Ordering::Relaxed);
355                async move {
356                    sleep(Duration::from_millis(5)).await;
357                    let out = Output { conn, resp };
358                    let resp = Response::new(Body::from(serde_json::to_vec(&out).unwrap()));
359                    Ok::<_, Infallible>(resp)
360                }
361            }))
362        })
363    }
364
365    #[test]
366    fn blocking_client_drives_the_composed_http_stack() {
367        let client = EasyHttpWebClient::connector_builder()
368            .with_custom_transport_connector(dummy_server())
369            .without_dns_connector()
370            .without_tls_proxy_support()
371            .without_proxy_support()
372            .without_tls_support()
373            .with_default_http_connector(Executor::default())
374            .without_connection_pool()
375            .build_client()
376            .try_into_blocking()
377            .unwrap();
378
379        let cloned = client.clone();
380        drop(client);
381        let response = cloned.get("http://example.com").send().unwrap();
382        assert_eq!(
383            response.try_into_json::<Output>().unwrap(),
384            Output { conn: 0, resp: 0 }
385        );
386    }
387
388    #[test]
389    fn default_blocking_http_client_is_cloneable_and_pooled() {
390        fn assert_default_client(_: &DefaultHttpWebClient) {}
391
392        let client = EasyHttpWebClient::try_blocking().unwrap();
393        assert_default_client(client.get_ref());
394        let cloned = client.clone();
395        drop(client);
396        let request = cloned.get("https://example.com").build().unwrap();
397        assert_eq!(request.uri(), &"https://example.com".parse().unwrap());
398    }
399
400    #[cfg(feature = "ws")]
401    #[test]
402    fn default_blocking_http_client_builds_websocket_requests() {
403        use crate::http::ws::handshake::client::BlockingHttpClientWebSocketExt as _;
404
405        let client = EasyHttpWebClient::try_blocking().unwrap();
406        let _from_url = client
407            .websocket("wss://example.com/chat")
408            .with_header("authorization", "Bearer secret");
409
410        let request = Request::builder()
411            .uri("wss://example.com/chat")
412            .body(Body::empty())
413            .unwrap();
414        let _from_request = client.websocket_with_request(request);
415    }
416
417    #[tokio::test]
418    async fn no_pool_tries_proxy_routes_in_order() {
419        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
420        let transport = service_fn({
421            let attempts = attempts.clone();
422            let direct = dummy_server::<ConnectRequest>();
423            move |input: ConnectRequest| {
424                let attempts = attempts.clone();
425                let direct = direct.clone();
426                async move {
427                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
428                    attempts.lock().push(route.clone());
429                    if route.proxy_address().is_some() {
430                        Err(ConnectionError::transport(
431                            BoxError::from_static_str("proxy unavailable"),
432                            ConnectionErrorKind::Unavailable,
433                        ))
434                    } else {
435                        direct.connect(input).await
436                    }
437                }
438            }
439        });
440        let client = EasyHttpWebClient::connector_builder()
441            .with_custom_transport_connector(transport)
442            .without_dns_connector()
443            .without_tls_proxy_support()
444            .with_custom_proxy_connector(())
445            .without_tls_support()
446            .with_default_http_connector(Executor::default())
447            .without_connection_pool()
448            .build_client();
449        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse::<ProxyAddress>().unwrap());
450        let request = || {
451            let request = Request::builder()
452                .uri("http://example.com")
453                .body(Body::empty())
454                .unwrap();
455            request
456                .extensions()
457                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
458            request
459        };
460
461        for _ in 0..2 {
462            client
463                .serve(request())
464                .await
465                .context("serve request through direct fallback")
466                .unwrap();
467        }
468
469        assert_eq!(
470            attempts.lock().as_slice(),
471            [proxy, ProxyRoute::Direct, ProxyRoute::Direct]
472        );
473    }
474
475    #[tokio::test]
476    async fn no_proxy_tls_support_falls_back_from_https_proxy() {
477        let client = EasyHttpWebClient::connector_builder()
478            .with_custom_transport_connector(dummy_server())
479            .without_dns_connector()
480            .without_tls_proxy_support()
481            .with_proxy_support()
482            .without_tls_support()
483            .with_default_http_connector(Executor::default())
484            .without_connection_pool()
485            .build_client();
486        let request = Request::builder()
487            .uri("http://example.com")
488            .body(Body::empty())
489            .unwrap();
490        request.extensions().insert(ProxyRoutes::new([
491            ProxyRoute::Proxy(
492                "https://proxy.example:8443"
493                    .parse::<ProxyAddress>()
494                    .unwrap(),
495            ),
496            ProxyRoute::Direct,
497        ]));
498
499        let response = client.serve(request).await.unwrap();
500        let output = response.try_into_json::<Output>().await.unwrap();
501        assert_eq!(output.conn, 0);
502        assert_eq!(output.resp, 0);
503    }
504
505    #[cfg(feature = "socks5")]
506    #[tokio::test]
507    async fn umbrella_proxy_connector_falls_back_across_mixed_plan() {
508        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
509        let transport = service_fn({
510            let attempts = attempts.clone();
511            let direct = dummy_server::<ConnectRequest>();
512            move |input: ConnectRequest| {
513                let attempts = attempts.clone();
514                let direct = direct.clone();
515                async move {
516                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap().clone();
517                    attempts.lock().push(route.clone());
518                    if route.proxy_address().is_some() {
519                        Err(ConnectionError::transport(
520                            BoxError::from_static_str("proxy unavailable"),
521                            ConnectionErrorKind::Unavailable,
522                        ))
523                    } else {
524                        direct.connect(input).await
525                    }
526                }
527            }
528        });
529        let client = EasyHttpWebClient::connector_builder()
530            .with_custom_transport_connector(transport)
531            .without_dns_connector()
532            .without_tls_proxy_support()
533            .with_proxy_support()
534            .without_tls_support()
535            .with_default_http_connector(Executor::default())
536            .without_connection_pool()
537            .build_client();
538        let request = Request::builder()
539            .uri("http://example.com")
540            .body(Body::empty())
541            .unwrap();
542        let unsupported = ProxyRoute::Proxy(
543            "custom-proxy://unsupported.example:8080"
544                .parse::<ProxyAddress>()
545                .unwrap(),
546        );
547        let socks = ProxyRoute::Proxy(
548            "socks5://socks.example:1080"
549                .parse::<ProxyAddress>()
550                .unwrap(),
551        );
552        let http = ProxyRoute::Proxy("http://http.example:8080".parse::<ProxyAddress>().unwrap());
553        request.extensions().insert(ProxyRoutes::new([
554            unsupported,
555            socks.clone(),
556            http.clone(),
557            ProxyRoute::Direct,
558        ]));
559
560        let response = client.serve(request).await.unwrap();
561        let output = response.try_into_json::<Output>().await.unwrap();
562        assert_eq!(output, Output { conn: 0, resp: 0 });
563        // The unsupported route is rejected by the umbrella dispatcher before
564        // transport. Reaching these three attempts proves that rejection was
565        // classified as retryable.
566        assert_eq!(
567            attempts.lock().as_slice(),
568            [socks, http, ProxyRoute::Direct]
569        );
570    }
571
572    #[tokio::test]
573    async fn default_pool_caches_failed_route_and_reuses_selected_connection() {
574        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
575        let transport = service_fn({
576            let attempts = attempts.clone();
577            let direct = dummy_server::<ConnectRequest>();
578            move |input: ConnectRequest| {
579                let attempts = attempts.clone();
580                let direct = direct.clone();
581                async move {
582                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
583                    attempts.lock().push(route.clone());
584                    if route.proxy_address().is_some() {
585                        Err(ConnectionError::transport(
586                            BoxError::from_static_str("proxy unavailable"),
587                            ConnectionErrorKind::Unavailable,
588                        ))
589                    } else {
590                        direct.connect(input).await
591                    }
592                }
593            }
594        });
595        let client = EasyHttpWebClient::connector_builder()
596            .with_custom_transport_connector(transport)
597            .without_dns_connector()
598            .without_tls_proxy_support()
599            .with_custom_proxy_connector(())
600            .without_tls_support()
601            .with_default_http_connector(Executor::default())
602            .with_default_connection_pool()
603            .build_client();
604        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse::<ProxyAddress>().unwrap());
605        let request = || {
606            let request = Request::builder()
607                .uri("http://example.com")
608                .body(Body::empty())
609                .unwrap();
610            request
611                .extensions()
612                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
613            request
614        };
615
616        for expected_response_index in 0..2 {
617            let response = client.serve(request()).await.unwrap();
618            let output = response.try_into_json::<Output>().await.unwrap();
619            assert_eq!(output.conn, 0);
620            assert_eq!(output.resp, expected_response_index);
621        }
622
623        assert_eq!(attempts.lock().as_slice(), [proxy, ProxyRoute::Direct]);
624    }
625
626    #[tokio::test]
627    async fn easy_client_can_disable_proxy_route_failure_cache() {
628        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
629        let transport = service_fn({
630            let attempts = attempts.clone();
631            let direct = dummy_server::<ConnectRequest>();
632            move |input: ConnectRequest| {
633                let attempts = attempts.clone();
634                let direct = direct.clone();
635                async move {
636                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
637                    attempts.lock().push(route.clone());
638                    if route.proxy_address().is_some() {
639                        Err(ConnectionError::transport(
640                            BoxError::from_static_str("proxy unavailable"),
641                            ConnectionErrorKind::Unavailable,
642                        ))
643                    } else {
644                        direct.connect(input).await
645                    }
646                }
647            }
648        });
649        let client = EasyHttpWebClient::connector_builder()
650            .with_custom_transport_connector(transport)
651            .without_dns_connector()
652            .without_tls_proxy_support()
653            .with_custom_proxy_connector(())
654            .without_tls_support()
655            .with_default_http_connector(Executor::default())
656            .without_proxy_route_failure_cache()
657            .without_connection_pool()
658            .build_client();
659        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse().unwrap());
660
661        for _ in 0..2 {
662            let request = Request::builder()
663                .uri("http://example.com")
664                .body(Body::empty())
665                .unwrap();
666            request
667                .extensions()
668                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
669            client.serve(request).await.unwrap();
670        }
671
672        assert_eq!(
673            attempts.lock().as_slice(),
674            [proxy.clone(), ProxyRoute::Direct, proxy, ProxyRoute::Direct]
675        );
676    }
677
678    #[tokio::test]
679    async fn proxy_free_easy_client_omits_proxy_route_failure_cache() {
680        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
681        let transport = service_fn({
682            let attempts = attempts.clone();
683            let direct = dummy_server::<ConnectRequest>();
684            move |input: ConnectRequest| {
685                let attempts = attempts.clone();
686                let direct = direct.clone();
687                async move {
688                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
689                    attempts.lock().push(route.clone());
690                    if route.proxy_address().is_some() {
691                        Err(ConnectionError::transport(
692                            BoxError::from_static_str("proxy unavailable"),
693                            ConnectionErrorKind::Unavailable,
694                        ))
695                    } else {
696                        direct.connect(input).await
697                    }
698                }
699            }
700        });
701        let client = EasyHttpWebClient::connector_builder()
702            .with_custom_transport_connector(transport)
703            .without_dns_connector()
704            .without_tls_proxy_support()
705            .without_proxy_support()
706            .without_tls_support()
707            .with_default_http_connector(Executor::default())
708            .without_connection_pool()
709            .build_client();
710        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse().unwrap());
711
712        for _ in 0..2 {
713            let request = Request::builder()
714                .uri("http://example.com")
715                .body(Body::empty())
716                .unwrap();
717            request
718                .extensions()
719                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
720            client.serve(request).await.unwrap();
721        }
722
723        assert_eq!(
724            attempts.lock().as_slice(),
725            [proxy.clone(), ProxyRoute::Direct, proxy, ProxyRoute::Direct]
726        );
727    }
728
729    #[tokio::test]
730    async fn easy_client_accepts_custom_proxy_route_failure_cache() {
731        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
732        let transport = service_fn({
733            let attempts = attempts.clone();
734            let direct = dummy_server::<ConnectRequest>();
735            move |input: ConnectRequest| {
736                let attempts = attempts.clone();
737                let direct = direct.clone();
738                async move {
739                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
740                    attempts.lock().push(route.clone());
741                    if route.proxy_address().is_some() {
742                        Err(ConnectionError::transport(
743                            BoxError::from_static_str("proxy unavailable"),
744                            ConnectionErrorKind::Unavailable,
745                        ))
746                    } else {
747                        direct.connect(input).await
748                    }
749                }
750            }
751        });
752        let mut failure_cache_config = ProxyRouteFailureCacheConfig::default();
753        failure_cache_config.scope = ProxyRouteFailureCacheScope::PerProxy;
754        let failure_cache = ProxyRouteFailureCache::try_new(failure_cache_config).unwrap();
755        let client = EasyHttpWebClient::connector_builder()
756            .with_custom_transport_connector(transport)
757            .without_dns_connector()
758            .without_tls_proxy_support()
759            .without_proxy_support()
760            .without_tls_support()
761            .with_default_http_connector(Executor::default())
762            .with_proxy_route_failure_cache(failure_cache)
763            .without_connection_pool()
764            .build_client();
765        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse().unwrap());
766
767        for destination in ["one.example", "two.example"] {
768            let request = Request::builder()
769                .uri(format!("http://{destination}"))
770                .body(Body::empty())
771                .unwrap();
772            request
773                .extensions()
774                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
775            client.serve(request).await.unwrap();
776        }
777
778        assert_eq!(
779            attempts.lock().as_slice(),
780            [proxy, ProxyRoute::Direct, ProxyRoute::Direct]
781        );
782    }
783
784    #[cfg(feature = "boring")]
785    #[test]
786    fn proxy_failure_cache_keeps_tls_client_future_bounded() {
787        let client = EasyHttpWebClient::connector_builder()
788            .with_default_transport_connector()
789            .with_default_dns_connector()
790            .without_tls_proxy_support()
791            .with_proxy_support()
792            .with_tls_support_using_boringssl_and_default_http_version(
793                crate::tls::client::TlsClientConfig::default_http(),
794                Version::HTTP_11,
795            )
796            .with_default_http_connector(Executor::default())
797            .without_connection_pool()
798            .build_client();
799        let request = Request::builder()
800            .uri("https://example.com")
801            .body(Body::empty())
802            .unwrap();
803
804        let future = client.serve(request);
805        let future_size = std::mem::size_of_val(&future);
806
807        assert!(
808            future_size < 64 * 1024,
809            "easy TLS client future is unexpectedly large: {future_size} bytes"
810        );
811    }
812
813    #[tokio::test]
814    async fn connection_is_in_use_until_response_body_is_consumed() {
815        let client = EasyHttpWebClient::connector_builder()
816            .with_custom_transport_connector(dummy_server())
817            .without_dns_connector()
818            .without_tls_proxy_support()
819            .without_proxy_support()
820            .without_tls_support()
821            .with_default_http_connector(Executor::default())
822            .try_with_connection_pool(HttpPooledConnectorConfig {
823                max_concurrent_streams: 1,
824                max_total: 4,
825                ..Default::default()
826            })
827            .unwrap()
828            .build_client();
829
830        let req = || {
831            Request::builder()
832                .uri("http://example.com")
833                .version(Version::HTTP_2)
834                .body(Body::empty())
835                .unwrap()
836        };
837
838        // Get the first response but DO NOT consume its body yet: the connection
839        // is logically still in use until the body is drained. Then issue a second
840        // request before draining the first.
841        let res1 = client.serve(req()).await.unwrap();
842        let res2 = client.serve(req()).await.unwrap();
843
844        // Drain in reverse so `res1`'s body is still outstanding when `req2` runs.
845        let out2 = res2.try_into_json::<Output>().await.unwrap();
846        let out1 = res1.try_into_json::<Output>().await.unwrap();
847
848        assert_eq!(out1.conn, 0, "first request uses the first connection");
849        // With `max_concurrent_streams = 1`, connection 0's response body is still
850        // in flight, so the second request must NOT reuse it.
851        assert_eq!(
852            out2.conn, 1,
853            "second request must not reuse a connection whose response body is still in flight"
854        );
855    }
856
857    // These things are already tested inside the pool itself, but here we add some high level tests
858    // in case we ever swap the underlying pool implementation.
859
860    #[tokio::test]
861    async fn default_pool_multiplexes_on_h2() {
862        let client = EasyHttpWebClient::connector_builder()
863            .with_custom_transport_connector(dummy_server())
864            .without_dns_connector()
865            .without_tls_proxy_support()
866            .without_proxy_support()
867            .without_tls_support()
868            .with_default_http_connector(Executor::default())
869            .with_default_connection_pool()
870            .build_client();
871
872        let req = || {
873            Request::builder()
874                .uri("http://example.com")
875                .version(Version::HTTP_2)
876                .body(Body::empty())
877                .unwrap()
878        };
879        let (res1, res2, res3) = tokio::join!(
880            client.serve(req()),
881            client.serve(req()),
882            client.serve(req()),
883        );
884
885        // Should only create single connection and send all requests over the same one
886        for (i, res) in [res1, res2, res3].into_iter().enumerate() {
887            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
888            assert_eq!(out.conn, 0);
889            assert_eq!(out.resp, i);
890        }
891    }
892
893    #[tokio::test]
894    async fn default_pool_does_not_multiplexes_on_h1() {
895        let client = EasyHttpWebClient::connector_builder()
896            .with_custom_transport_connector(dummy_server())
897            .without_dns_connector()
898            .without_tls_proxy_support()
899            .without_proxy_support()
900            .without_tls_support()
901            .with_default_http_connector(Executor::default())
902            .with_default_connection_pool()
903            .build_client();
904
905        let req = || {
906            Request::builder()
907                .uri("http://example.com")
908                .version(Version::HTTP_11)
909                .body(Body::empty())
910                .unwrap()
911        };
912        let (res1, res2, res3) = tokio::join!(
913            client.serve(req()),
914            client.serve(req()),
915            client.serve(req()),
916        );
917
918        // Should create a new connection for each request since they are all inprogress at the same
919        // time and h1 does not support multiplexing
920        for (i, res) in [res1, res2, res3].into_iter().enumerate() {
921            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
922            assert_eq!(out.conn, i);
923            assert_eq!(out.resp, 0);
924        }
925    }
926
927    #[tokio::test]
928    async fn multiplex_on_h2_respects_limits() {
929        let client = EasyHttpWebClient::connector_builder()
930            .with_custom_transport_connector(dummy_server())
931            .without_dns_connector()
932            .without_tls_proxy_support()
933            .without_proxy_support()
934            .without_tls_support()
935            .with_default_http_connector(Executor::default())
936            .try_with_connection_pool(HttpPooledConnectorConfig {
937                max_concurrent_streams: 2,
938                ..Default::default()
939            })
940            .unwrap()
941            .build_client();
942
943        let req = || {
944            Request::builder()
945                .uri("http://example.com")
946                .version(Version::HTTP_2)
947                .body(Body::empty())
948                .unwrap()
949        };
950        let (res1, res2, res3, res4) = tokio::join!(
951            client.serve(req()),
952            client.serve(req()),
953            client.serve(req()),
954            client.serve(req()),
955        );
956
957        // Should create a connection for every two request
958        for (i, res) in [res1, res2, res3, res4].into_iter().enumerate() {
959            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
960            assert_eq!(out.conn, i / 2);
961            assert_eq!(out.resp, i % 2);
962        }
963    }
964}
rama/http/client/mod.rs

rama/http/client/
mod.rs
