rama 0.4.0 - Docs.rs

//! rama http client support
//!
//! Contains re-exports from `rama-http-backend::client`
//! and adds `EasyHttpWebClient`, an opiniated http web client which
//! supports most common use cases and provides sensible defaults.
use std::{fmt, io};

use crate::{
    Layer, Service,
    error::BoxError,
    extensions::ExtensionsRef,
    http::{Request, Response, StreamingBody},
    net::client::EstablishedClientConnection,
    rt::Executor,
    service::BoxService,
    telemetry::tracing,
};

#[doc(inline)]
pub use ::rama_http::service::client::blocking::{
    Body as BlockingBody, Client as BlockingHttpClient, Response as BlockingResponse,
};
#[doc(inline)]
pub use ::rama_http_backend::client::*;
use rama_core::{
    error::{ErrorContext, ErrorExt as _, extra::OpaqueError},
    extensions::Egress,
    layer::MapErr,
};

pub mod builder;
#[doc(inline)]
pub use builder::EasyHttpConnectorBuilder;

#[cfg(feature = "socks5")]
mod proxy_connector;
#[cfg(feature = "socks5")]
#[cfg_attr(docsrs, doc(cfg(feature = "socks5")))]
#[doc(inline)]
pub use proxy_connector::{MaybeProxiedConnection, ProxyConnector, ProxyConnectorLayer};

/// An opiniated http client that can be used to serve HTTP requests.
///
/// Use [`EasyHttpWebClient::connector_builder()`] to easily create a client with
/// a common Http connector setup (tcp + proxy + tls + http) or bring your
/// own http connector.
///
/// [`Default`] uses Rama's default multiplexing connection pool. Build the
/// connector explicitly with
/// [`EasyHttpConnectorBuilder::without_connection_pool`] when connection reuse
/// is unwanted.
///
/// You can fork this http client in case you have use cases not possible with this service example.
/// E.g. perhaps you wish to have middleware in into outbound requests, after they
/// passed through your "connector" setup. All this and more is possible by defining your own
/// http client. Rama is here to empower you, the building blocks are there, go crazy
/// with your own service fork and use the full power of Rust at your fingertips ;)
pub struct EasyHttpWebClient<BodyIn, ConnResponse, L> {
    connector: BoxService<Request<BodyIn>, ConnResponse, OpaqueError>,
    jit_layers: L,
}

impl<BodyIn, ConnResponse, L> fmt::Debug for EasyHttpWebClient<BodyIn, ConnResponse, L> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("EasyHttpWebClient").finish()
    }
}

impl<BodyIn, ConnResponse, L: Clone> Clone for EasyHttpWebClient<BodyIn, ConnResponse, L> {
    fn clone(&self) -> Self {
        Self {
            connector: self.connector.clone(),
            jit_layers: self.jit_layers.clone(),
        }
    }
}

impl EasyHttpWebClient<(), (), ()> {
    /// Create a [`EasyHttpConnectorBuilder`] to easily create a [`EasyHttpWebClient`] with a custom connector
    #[must_use]
    pub fn connector_builder() -> EasyHttpConnectorBuilder {
        EasyHttpConnectorBuilder::new()
    }

    /// Create a cloneable blocking HTTP(S) client with its own dedicated
    /// runtime thread and Rama's default web connector stack.
    ///
    /// ```no_run
    /// use rama::http::client::EasyHttpWebClient;
    ///
    /// # fn main() -> Result<(), rama::error::BoxError> {
    /// let client = EasyHttpWebClient::try_blocking()?;
    /// let client_for_worker = client.clone();
    ///
    /// let text = client_for_worker
    ///     .get("https://example.com/")
    ///     .send()?
    ///     .try_into_string()?;
    /// # _ = text;
    /// # Ok(())
    /// # }
    /// ```
    pub fn try_blocking() -> io::Result<BlockingHttpWebClient> {
        BlockingHttpClient::try_new(EasyHttpWebClient::default())
    }
}

/// Rama's default asynchronous HTTP(S) client, including its default
/// multiplexing connection pool.
pub type DefaultHttpWebClient<Body = crate::http::Body> = EasyHttpWebClient<
    Body,
    EstablishedClientConnection<
        BindBodyToConn<
            crate::net::client::pool::MultiplexedConnection<
                HttpClientService<Body>,
                BasicHttpConId,
            >,
        >,
        Request<Body>,
    >,
    (),
>;

/// A blocking HTTP(S) client using Rama's default pooled web connector stack.
pub type BlockingHttpWebClient = BlockingHttpClient<DefaultHttpWebClient>;

impl<Body> Default for DefaultHttpWebClient<Body>
where
    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
{
    #[inline(always)]
    fn default() -> Self {
        Self::default_with_executor(Executor::default())
    }
}

impl<Body> DefaultHttpWebClient<Body>
where
    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
{
    core::cfg_select! {
        feature = "boring" => {
            pub fn default_with_executor(exec: Executor) -> Self {
                let tls_config = crate::tls::client::TlsClientConfig::default_http();

                EasyHttpConnectorBuilder::new()
                    .with_default_transport_connector()
                    .with_default_dns_connector()
                    .with_tls_proxy_support_using_boringssl()
                    .with_proxy_support()
                    .with_tls_support_using_boringssl(tls_config)
                    .with_default_http_connector(exec)
                    .with_default_connection_pool()
                    .build_client()
            }
        }
        feature = "rustls" => {
            pub fn default_with_executor(exec: Executor) -> Self {
                let tls_config = crate::tls::client::TlsClientConfig::default_http();

                EasyHttpConnectorBuilder::new()
                    .with_default_transport_connector()
                    .with_default_dns_connector()
                    .with_tls_proxy_support_using_rustls()
                    .with_proxy_support()
                    .with_tls_support_using_rustls(tls_config)
                    .with_default_http_connector(exec)
                    .with_default_connection_pool()
                    .build_client()
            }
        }
        _ => {
            pub fn default_with_executor(exec: Executor) -> Self {
                EasyHttpConnectorBuilder::new()
                    .with_default_transport_connector()
                    .with_default_dns_connector()
                    .without_tls_proxy_support()
                    .with_proxy_support()
                    .without_tls_support()
                    .with_default_http_connector(exec)
                    .with_default_connection_pool()
                    .build_client()
            }
        }
    }
}

impl<BodyIn, ConnResponse> EasyHttpWebClient<BodyIn, ConnResponse, ()>
where
    BodyIn: Send + 'static,
{
    /// Create a new [`EasyHttpWebClient`] using the provided connector
    #[must_use]
    pub fn new<S>(connector: S) -> Self
    where
        S: Service<Request<BodyIn>, Output = ConnResponse, Error: Into<BoxError>>,
    {
        Self {
            connector: MapErr::into_opaque_error(connector).boxed(),
            jit_layers: (),
        }
    }
}

impl<BodyIn, ConnResponse, L> EasyHttpWebClient<BodyIn, ConnResponse, L> {
    /// Convert this asynchronous web client into a cloneable blocking client
    /// with its own dedicated runtime thread.
    pub fn try_into_blocking(self) -> io::Result<BlockingHttpClient<Self>> {
        BlockingHttpClient::try_new(self)
    }

    /// Convert this asynchronous web client into a blocking client using a
    /// caller-supplied runtime.
    #[must_use]
    pub fn into_blocking_with_runtime(
        self,
        runtime: &crate::rt::blocking::Runtime,
    ) -> BlockingHttpClient<Self> {
        BlockingHttpClient::with_runtime(self, runtime)
    }

    /// Set the connector that this [`EasyHttpWebClient`] will use
    #[must_use]
    pub fn with_connector<S, BodyInNew, ConnResponseNew>(
        self,
        connector: S,
    ) -> EasyHttpWebClient<BodyInNew, ConnResponseNew, L>
    where
        S: Service<Request<BodyInNew>, Output = ConnResponseNew, Error: Into<BoxError>>,
        BodyInNew: Send + 'static,
    {
        EasyHttpWebClient {
            connector: MapErr::into_opaque_error(connector).boxed(),
            jit_layers: self.jit_layers,
        }
    }

    /// [`Layer`] which will be applied just in time (JIT) before the request is send, but after
    /// the connection has been established.
    ///
    /// Simplified flow of how the [`EasyHttpWebClient`] works:
    /// 1. External: let response = client.serve(request)
    /// 2. Internal: let http_connection = self.connector.serve(request)
    /// 3. Internal: let response = jit_layers.layer(http_connection).serve(request)
    pub fn with_jit_layer<T>(self, jit_layers: T) -> EasyHttpWebClient<BodyIn, ConnResponse, T> {
        EasyHttpWebClient {
            connector: self.connector,
            jit_layers,
        }
    }
}

impl<Body, ConnectionBody, Connection, L> Service<Request<Body>>
    for EasyHttpWebClient<Body, EstablishedClientConnection<Connection, Request<ConnectionBody>>, L>
where
    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
    Connection:
        Service<Request<ConnectionBody>, Output = Response, Error = BoxError> + ExtensionsRef,
    // Body type this connection will be able to send, this is not necessarily the same one that
    // was used in the request that created this connection
    ConnectionBody:
        StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
    L: Layer<
            Connection,
            Service: Service<Request<ConnectionBody>, Output = Response, Error = BoxError>,
        > + Send
        + Sync
        + 'static,
{
    type Output = Response;
    type Error = OpaqueError;

    async fn serve(&self, req: Request<Body>) -> Result<Self::Output, Self::Error> {
        let uri = req.uri().clone();

        let EstablishedClientConnection {
            input: req,
            conn: http_connection,
        } = self.connector.serve(req).await.into_opaque_error()?;

        req.extensions()
            .insert(Egress(http_connection.extensions().clone()));

        let http_connection = self.jit_layers.layer(http_connection);

        // NOTE: stack might change request version based on connector data,
        tracing::trace!(url.full = %uri, "send http req to connector stack");

        let result = http_connection.serve(req).await;

        match result {
            Ok(resp) => {
                tracing::trace!(url.full = %uri, "response received from connector stack");
                Ok(resp)
            }
            Err(err) => Err(err
                .context("http request failure")
                .context_field("uri", uri)
                .into_opaque_error()),
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{
        convert::Infallible,
        sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        },
        time::Duration,
    };

    use rama_core::{error::BoxErrorExt as _, service::service_fn};
    use rama_http::{Body, BodyExtractExt, Version};
    use rama_http_backend::server::HttpServer;
    use rama_net::{
        address::ProxyAddress,
        client::{
            ConnectRequest, ConnectionError, ConnectionErrorKind, ConnectorService, ProxyRoute,
            ProxyRouteFailureCache, ProxyRouteFailureCacheConfig, ProxyRouteFailureCacheScope,
            ProxyRoutes,
        },
        test_utils::client::{MockConnectorService, MockSocket},
    };
    use serde::{Deserialize, Serialize};
    use tokio::time::sleep;

    use super::*;

    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
    struct Output {
        conn: usize,
        resp: usize,
    }

    fn dummy_server<Input: Send + 'static>()
    -> impl Service<
        Input,
        Output = EstablishedClientConnection<MockSocket, Input>,
        Error = Infallible,
    > + Clone {
        let created_connections = Arc::new(AtomicUsize::new(0));
        MockConnectorService::new(move || {
            let created_connections = created_connections.clone();
            let conn = created_connections.fetch_add(1, Ordering::Relaxed);

            // count responses created on this specific connection
            let created_response = Arc::new(AtomicUsize::new(0));

            HttpServer::auto(Executor::default()).service(service_fn(move |_req: Request| {
                let created_response = created_response.clone();
                let resp = created_response.fetch_add(1, Ordering::Relaxed);
                async move {
                    sleep(Duration::from_millis(5)).await;
                    let out = Output { conn, resp };
                    let resp = Response::new(Body::from(serde_json::to_vec(&out).unwrap()));
                    Ok::<_, Infallible>(resp)
                }
            }))
        })
    }

    #[test]
    fn blocking_client_drives_the_composed_http_stack() {
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(dummy_server())
            .without_dns_connector()
            .without_tls_proxy_support()
            .without_proxy_support()
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .without_connection_pool()
            .build_client()
            .try_into_blocking()
            .unwrap();

        let cloned = client.clone();
        drop(client);
        let response = cloned.get("http://example.com").send().unwrap();
        assert_eq!(
            response.try_into_json::<Output>().unwrap(),
            Output { conn: 0, resp: 0 }
        );
    }

    #[test]
    fn default_blocking_http_client_is_cloneable_and_pooled() {
        fn assert_default_client(_: &DefaultHttpWebClient) {}

        let client = EasyHttpWebClient::try_blocking().unwrap();
        assert_default_client(client.get_ref());
        let cloned = client.clone();
        drop(client);
        let request = cloned.get("https://example.com").build().unwrap();
        assert_eq!(request.uri(), &"https://example.com".parse().unwrap());
    }

    #[cfg(feature = "ws")]
    #[test]
    fn default_blocking_http_client_builds_websocket_requests() {
        use crate::http::ws::handshake::client::BlockingHttpClientWebSocketExt as _;

        let client = EasyHttpWebClient::try_blocking().unwrap();
        let _from_url = client
            .websocket("wss://example.com/chat")
            .with_header("authorization", "Bearer secret");

        let request = Request::builder()
            .uri("wss://example.com/chat")
            .body(Body::empty())
            .unwrap();
        let _from_request = client.websocket_with_request(request);
    }

    #[tokio::test]
    async fn no_pool_tries_proxy_routes_in_order() {
        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
        let transport = service_fn({
            let attempts = attempts.clone();
            let direct = dummy_server::<ConnectRequest>();
            move |input: ConnectRequest| {
                let attempts = attempts.clone();
                let direct = direct.clone();
                async move {
                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
                    attempts.lock().push(route.clone());
                    if route.proxy_address().is_some() {
                        Err(ConnectionError::transport(
                            BoxError::from_static_str("proxy unavailable"),
                            ConnectionErrorKind::Unavailable,
                        ))
                    } else {
                        direct.connect(input).await
                    }
                }
            }
        });
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(transport)
            .without_dns_connector()
            .without_tls_proxy_support()
            .with_custom_proxy_connector(())
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .without_connection_pool()
            .build_client();
        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse::<ProxyAddress>().unwrap());
        let request = || {
            let request = Request::builder()
                .uri("http://example.com")
                .body(Body::empty())
                .unwrap();
            request
                .extensions()
                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
            request
        };

        for _ in 0..2 {
            client
                .serve(request())
                .await
                .context("serve request through direct fallback")
                .unwrap();
        }

        assert_eq!(
            attempts.lock().as_slice(),
            [proxy, ProxyRoute::Direct, ProxyRoute::Direct]
        );
    }

    #[tokio::test]
    async fn no_proxy_tls_support_falls_back_from_https_proxy() {
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(dummy_server())
            .without_dns_connector()
            .without_tls_proxy_support()
            .with_proxy_support()
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .without_connection_pool()
            .build_client();
        let request = Request::builder()
            .uri("http://example.com")
            .body(Body::empty())
            .unwrap();
        request.extensions().insert(ProxyRoutes::new([
            ProxyRoute::Proxy(
                "https://proxy.example:8443"
                    .parse::<ProxyAddress>()
                    .unwrap(),
            ),
            ProxyRoute::Direct,
        ]));

        let response = client.serve(request).await.unwrap();
        let output = response.try_into_json::<Output>().await.unwrap();
        assert_eq!(output.conn, 0);
        assert_eq!(output.resp, 0);
    }

    #[cfg(feature = "socks5")]
    #[tokio::test]
    async fn umbrella_proxy_connector_falls_back_across_mixed_plan() {
        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
        let transport = service_fn({
            let attempts = attempts.clone();
            let direct = dummy_server::<ConnectRequest>();
            move |input: ConnectRequest| {
                let attempts = attempts.clone();
                let direct = direct.clone();
                async move {
                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap().clone();
                    attempts.lock().push(route.clone());
                    if route.proxy_address().is_some() {
                        Err(ConnectionError::transport(
                            BoxError::from_static_str("proxy unavailable"),
                            ConnectionErrorKind::Unavailable,
                        ))
                    } else {
                        direct.connect(input).await
                    }
                }
            }
        });
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(transport)
            .without_dns_connector()
            .without_tls_proxy_support()
            .with_proxy_support()
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .without_connection_pool()
            .build_client();
        let request = Request::builder()
            .uri("http://example.com")
            .body(Body::empty())
            .unwrap();
        let unsupported = ProxyRoute::Proxy(
            "custom-proxy://unsupported.example:8080"
                .parse::<ProxyAddress>()
                .unwrap(),
        );
        let socks = ProxyRoute::Proxy(
            "socks5://socks.example:1080"
                .parse::<ProxyAddress>()
                .unwrap(),
        );
        let http = ProxyRoute::Proxy("http://http.example:8080".parse::<ProxyAddress>().unwrap());
        request.extensions().insert(ProxyRoutes::new([
            unsupported,
            socks.clone(),
            http.clone(),
            ProxyRoute::Direct,
        ]));

        let response = client.serve(request).await.unwrap();
        let output = response.try_into_json::<Output>().await.unwrap();
        assert_eq!(output, Output { conn: 0, resp: 0 });
        // The unsupported route is rejected by the umbrella dispatcher before
        // transport. Reaching these three attempts proves that rejection was
        // classified as retryable.
        assert_eq!(
            attempts.lock().as_slice(),
            [socks, http, ProxyRoute::Direct]
        );
    }

    #[tokio::test]
    async fn default_pool_caches_failed_route_and_reuses_selected_connection() {
        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
        let transport = service_fn({
            let attempts = attempts.clone();
            let direct = dummy_server::<ConnectRequest>();
            move |input: ConnectRequest| {
                let attempts = attempts.clone();
                let direct = direct.clone();
                async move {
                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
                    attempts.lock().push(route.clone());
                    if route.proxy_address().is_some() {
                        Err(ConnectionError::transport(
                            BoxError::from_static_str("proxy unavailable"),
                            ConnectionErrorKind::Unavailable,
                        ))
                    } else {
                        direct.connect(input).await
                    }
                }
            }
        });
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(transport)
            .without_dns_connector()
            .without_tls_proxy_support()
            .with_custom_proxy_connector(())
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .with_default_connection_pool()
            .build_client();
        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse::<ProxyAddress>().unwrap());
        let request = || {
            let request = Request::builder()
                .uri("http://example.com")
                .body(Body::empty())
                .unwrap();
            request
                .extensions()
                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
            request
        };

        for expected_response_index in 0..2 {
            let response = client.serve(request()).await.unwrap();
            let output = response.try_into_json::<Output>().await.unwrap();
            assert_eq!(output.conn, 0);
            assert_eq!(output.resp, expected_response_index);
        }

        assert_eq!(attempts.lock().as_slice(), [proxy, ProxyRoute::Direct]);
    }

    #[tokio::test]
    async fn easy_client_can_disable_proxy_route_failure_cache() {
        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
        let transport = service_fn({
            let attempts = attempts.clone();
            let direct = dummy_server::<ConnectRequest>();
            move |input: ConnectRequest| {
                let attempts = attempts.clone();
                let direct = direct.clone();
                async move {
                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
                    attempts.lock().push(route.clone());
                    if route.proxy_address().is_some() {
                        Err(ConnectionError::transport(
                            BoxError::from_static_str("proxy unavailable"),
                            ConnectionErrorKind::Unavailable,
                        ))
                    } else {
                        direct.connect(input).await
                    }
                }
            }
        });
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(transport)
            .without_dns_connector()
            .without_tls_proxy_support()
            .with_custom_proxy_connector(())
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .without_proxy_route_failure_cache()
            .without_connection_pool()
            .build_client();
        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse().unwrap());

        for _ in 0..2 {
            let request = Request::builder()
                .uri("http://example.com")
                .body(Body::empty())
                .unwrap();
            request
                .extensions()
                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
            client.serve(request).await.unwrap();
        }

        assert_eq!(
            attempts.lock().as_slice(),
            [proxy.clone(), ProxyRoute::Direct, proxy, ProxyRoute::Direct]
        );
    }

    #[tokio::test]
    async fn proxy_free_easy_client_omits_proxy_route_failure_cache() {
        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
        let transport = service_fn({
            let attempts = attempts.clone();
            let direct = dummy_server::<ConnectRequest>();
            move |input: ConnectRequest| {
                let attempts = attempts.clone();
                let direct = direct.clone();
                async move {
                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
                    attempts.lock().push(route.clone());
                    if route.proxy_address().is_some() {
                        Err(ConnectionError::transport(
                            BoxError::from_static_str("proxy unavailable"),
                            ConnectionErrorKind::Unavailable,
                        ))
                    } else {
                        direct.connect(input).await
                    }
                }
            }
        });
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(transport)
            .without_dns_connector()
            .without_tls_proxy_support()
            .without_proxy_support()
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .without_connection_pool()
            .build_client();
        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse().unwrap());

        for _ in 0..2 {
            let request = Request::builder()
                .uri("http://example.com")
                .body(Body::empty())
                .unwrap();
            request
                .extensions()
                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
            client.serve(request).await.unwrap();
        }

        assert_eq!(
            attempts.lock().as_slice(),
            [proxy.clone(), ProxyRoute::Direct, proxy, ProxyRoute::Direct]
        );
    }

    #[tokio::test]
    async fn easy_client_accepts_custom_proxy_route_failure_cache() {
        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
        let transport = service_fn({
            let attempts = attempts.clone();
            let direct = dummy_server::<ConnectRequest>();
            move |input: ConnectRequest| {
                let attempts = attempts.clone();
                let direct = direct.clone();
                async move {
                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
                    attempts.lock().push(route.clone());
                    if route.proxy_address().is_some() {
                        Err(ConnectionError::transport(
                            BoxError::from_static_str("proxy unavailable"),
                            ConnectionErrorKind::Unavailable,
                        ))
                    } else {
                        direct.connect(input).await
                    }
                }
            }
        });
        let mut failure_cache_config = ProxyRouteFailureCacheConfig::default();
        failure_cache_config.scope = ProxyRouteFailureCacheScope::PerProxy;
        let failure_cache = ProxyRouteFailureCache::try_new(failure_cache_config).unwrap();
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(transport)
            .without_dns_connector()
            .without_tls_proxy_support()
            .without_proxy_support()
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .with_proxy_route_failure_cache(failure_cache)
            .without_connection_pool()
            .build_client();
        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse().unwrap());

        for destination in ["one.example", "two.example"] {
            let request = Request::builder()
                .uri(format!("http://{destination}"))
                .body(Body::empty())
                .unwrap();
            request
                .extensions()
                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
            client.serve(request).await.unwrap();
        }

        assert_eq!(
            attempts.lock().as_slice(),
            [proxy, ProxyRoute::Direct, ProxyRoute::Direct]
        );
    }

    #[cfg(feature = "boring")]
    #[test]
    fn proxy_failure_cache_keeps_tls_client_future_bounded() {
        let client = EasyHttpWebClient::connector_builder()
            .with_default_transport_connector()
            .with_default_dns_connector()
            .without_tls_proxy_support()
            .with_proxy_support()
            .with_tls_support_using_boringssl_and_default_http_version(
                crate::tls::client::TlsClientConfig::default_http(),
                Version::HTTP_11,
            )
            .with_default_http_connector(Executor::default())
            .without_connection_pool()
            .build_client();
        let request = Request::builder()
            .uri("https://example.com")
            .body(Body::empty())
            .unwrap();

        let future = client.serve(request);
        let future_size = std::mem::size_of_val(&future);

        assert!(
            future_size < 64 * 1024,
            "easy TLS client future is unexpectedly large: {future_size} bytes"
        );
    }

    #[tokio::test]
    async fn connection_is_in_use_until_response_body_is_consumed() {
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(dummy_server())
            .without_dns_connector()
            .without_tls_proxy_support()
            .without_proxy_support()
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .try_with_connection_pool(HttpPooledConnectorConfig {
                max_concurrent_streams: 1,
                max_total: 4,
                ..Default::default()
            })
            .unwrap()
            .build_client();

        let req = || {
            Request::builder()
                .uri("http://example.com")
                .version(Version::HTTP_2)
                .body(Body::empty())
                .unwrap()
        };

        // Get the first response but DO NOT consume its body yet: the connection
        // is logically still in use until the body is drained. Then issue a second
        // request before draining the first.
        let res1 = client.serve(req()).await.unwrap();
        let res2 = client.serve(req()).await.unwrap();

        // Drain in reverse so `res1`'s body is still outstanding when `req2` runs.
        let out2 = res2.try_into_json::<Output>().await.unwrap();
        let out1 = res1.try_into_json::<Output>().await.unwrap();

        assert_eq!(out1.conn, 0, "first request uses the first connection");
        // With `max_concurrent_streams = 1`, connection 0's response body is still
        // in flight, so the second request must NOT reuse it.
        assert_eq!(
            out2.conn, 1,
            "second request must not reuse a connection whose response body is still in flight"
        );
    }

    // These things are already tested inside the pool itself, but here we add some high level tests
    // in case we ever swap the underlying pool implementation.

    #[tokio::test]
    async fn default_pool_multiplexes_on_h2() {
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(dummy_server())
            .without_dns_connector()
            .without_tls_proxy_support()
            .without_proxy_support()
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .with_default_connection_pool()
            .build_client();

        let req = || {
            Request::builder()
                .uri("http://example.com")
                .version(Version::HTTP_2)
                .body(Body::empty())
                .unwrap()
        };
        let (res1, res2, res3) = tokio::join!(
            client.serve(req()),
            client.serve(req()),
            client.serve(req()),
        );

        // Should only create single connection and send all requests over the same one
        for (i, res) in [res1, res2, res3].into_iter().enumerate() {
            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
            assert_eq!(out.conn, 0);
            assert_eq!(out.resp, i);
        }
    }

    #[tokio::test]
    async fn default_pool_does_not_multiplexes_on_h1() {
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(dummy_server())
            .without_dns_connector()
            .without_tls_proxy_support()
            .without_proxy_support()
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .with_default_connection_pool()
            .build_client();

        let req = || {
            Request::builder()
                .uri("http://example.com")
                .version(Version::HTTP_11)
                .body(Body::empty())
                .unwrap()
        };
        let (res1, res2, res3) = tokio::join!(
            client.serve(req()),
            client.serve(req()),
            client.serve(req()),
        );

        // Should create a new connection for each request since they are all inprogress at the same
        // time and h1 does not support multiplexing
        for (i, res) in [res1, res2, res3].into_iter().enumerate() {
            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
            assert_eq!(out.conn, i);
            assert_eq!(out.resp, 0);
        }
    }

    #[tokio::test]
    async fn multiplex_on_h2_respects_limits() {
        let client = EasyHttpWebClient::connector_builder()
            .with_custom_transport_connector(dummy_server())
            .without_dns_connector()
            .without_tls_proxy_support()
            .without_proxy_support()
            .without_tls_support()
            .with_default_http_connector(Executor::default())
            .try_with_connection_pool(HttpPooledConnectorConfig {
                max_concurrent_streams: 2,
                ..Default::default()
            })
            .unwrap()
            .build_client();

        let req = || {
            Request::builder()
                .uri("http://example.com")
                .version(Version::HTTP_2)
                .body(Body::empty())
                .unwrap()
        };
        let (res1, res2, res3, res4) = tokio::join!(
            client.serve(req()),
            client.serve(req()),
            client.serve(req()),
            client.serve(req()),
        );

        // Should create a connection for every two request
        for (i, res) in [res1, res2, res3, res4].into_iter().enumerate() {
            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
            assert_eq!(out.conn, i / 2);
            assert_eq!(out.resp, i % 2);
        }
    }
}