Skip to main content

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;
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_backend::client::*;
21use rama_core::{
22    error::{ErrorContext, ErrorExt as _, extra::OpaqueError},
23    extensions::Egress,
24    layer::MapErr,
25};
26
27pub mod builder;
28#[doc(inline)]
29pub use builder::EasyHttpConnectorBuilder;
30
31#[cfg(feature = "socks5")]
32mod proxy_connector;
33#[cfg(feature = "socks5")]
34#[cfg_attr(docsrs, doc(cfg(feature = "socks5")))]
35#[doc(inline)]
36pub use proxy_connector::{MaybeProxiedConnection, ProxyConnector, ProxyConnectorLayer};
37
38/// An opiniated http client that can be used to serve HTTP requests.
39///
40/// Use [`EasyHttpWebClient::connector_builder()`] to easily create a client with
41/// a common Http connector setup (tcp + proxy + tls + http) or bring your
42/// own http connector.
43///
44/// You can fork this http client in case you have use cases not possible with this service example.
45/// E.g. perhaps you wish to have middleware in into outbound requests, after they
46/// passed through your "connector" setup. All this and more is possible by defining your own
47/// http client. Rama is here to empower you, the building blocks are there, go crazy
48/// with your own service fork and use the full power of Rust at your fingertips ;)
49pub struct EasyHttpWebClient<BodyIn, ConnResponse, L> {
50    connector: BoxService<Request<BodyIn>, ConnResponse, OpaqueError>,
51    jit_layers: L,
52}
53
54impl<BodyIn, ConnResponse, L> fmt::Debug for EasyHttpWebClient<BodyIn, ConnResponse, L> {
55    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
56        f.debug_struct("EasyHttpWebClient").finish()
57    }
58}
59
60impl<BodyIn, ConnResponse, L: Clone> Clone for EasyHttpWebClient<BodyIn, ConnResponse, L> {
61    fn clone(&self) -> Self {
62        Self {
63            connector: self.connector.clone(),
64            jit_layers: self.jit_layers.clone(),
65        }
66    }
67}
68
69impl EasyHttpWebClient<(), (), ()> {
70    /// Create a [`EasyHttpConnectorBuilder`] to easily create a [`EasyHttpWebClient`] with a custom connector
71    #[must_use]
72    pub fn connector_builder() -> EasyHttpConnectorBuilder {
73        EasyHttpConnectorBuilder::new()
74    }
75}
76
77impl<Body> Default
78    for EasyHttpWebClient<
79        Body,
80        EstablishedClientConnection<HttpClientService<Body>, Request<Body>>,
81        (),
82    >
83where
84    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
85{
86    #[inline(always)]
87    fn default() -> Self {
88        Self::default_with_executor(Executor::default())
89    }
90}
91
92impl<Body>
93    EasyHttpWebClient<Body, EstablishedClientConnection<HttpClientService<Body>, Request<Body>>, ()>
94where
95    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
96{
97    core::cfg_select! {
98        feature = "boring" => {
99            pub fn default_with_executor(exec: Executor) -> Self {
100                let tls_config = crate::tls::client::TlsClientConfig::default_http();
101
102                EasyHttpConnectorBuilder::new()
103                    .with_default_transport_connector()
104                    .with_default_dns_connector()
105                    .with_tls_proxy_support_using_boringssl()
106                    .with_proxy_support()
107                    .with_tls_support_using_boringssl(tls_config)
108                    .with_default_http_connector(exec)
109                    .build_client()
110            }
111        }
112        feature = "rustls" => {
113            pub fn default_with_executor(exec: Executor) -> Self {
114                let tls_config = crate::tls::client::TlsClientConfig::default_http();
115
116                EasyHttpConnectorBuilder::new()
117                    .with_default_transport_connector()
118                    .with_default_dns_connector()
119                    .with_tls_proxy_support_using_rustls()
120                    .with_proxy_support()
121                    .with_tls_support_using_rustls(tls_config)
122                    .with_default_http_connector(exec)
123                    .build_client()
124            }
125        }
126        _ => {
127            pub fn default_with_executor(exec: Executor) -> Self {
128                EasyHttpConnectorBuilder::new()
129                    .with_default_transport_connector()
130                    .with_default_dns_connector()
131                    .without_tls_proxy_support()
132                    .with_proxy_support()
133                    .without_tls_support()
134                    .with_default_http_connector(exec)
135                    .build_client()
136            }
137        }
138    }
139}
140
141impl<BodyIn, ConnResponse> EasyHttpWebClient<BodyIn, ConnResponse, ()>
142where
143    BodyIn: Send + 'static,
144{
145    /// Create a new [`EasyHttpWebClient`] using the provided connector
146    #[must_use]
147    pub fn new<S>(connector: S) -> Self
148    where
149        S: Service<Request<BodyIn>, Output = ConnResponse, Error: Into<BoxError>>,
150    {
151        Self {
152            connector: MapErr::into_opaque_error(connector).boxed(),
153            jit_layers: (),
154        }
155    }
156}
157
158impl<BodyIn, ConnResponse, L> EasyHttpWebClient<BodyIn, ConnResponse, L> {
159    /// Set the connector that this [`EasyHttpWebClient`] will use
160    #[must_use]
161    pub fn with_connector<S, BodyInNew, ConnResponseNew>(
162        self,
163        connector: S,
164    ) -> EasyHttpWebClient<BodyInNew, ConnResponseNew, L>
165    where
166        S: Service<Request<BodyInNew>, Output = ConnResponseNew, Error: Into<BoxError>>,
167        BodyInNew: Send + 'static,
168    {
169        EasyHttpWebClient {
170            connector: MapErr::into_opaque_error(connector).boxed(),
171            jit_layers: self.jit_layers,
172        }
173    }
174
175    /// [`Layer`] which will be applied just in time (JIT) before the request is send, but after
176    /// the connection has been established.
177    ///
178    /// Simplified flow of how the [`EasyHttpWebClient`] works:
179    /// 1. External: let response = client.serve(request)
180    /// 2. Internal: let http_connection = self.connector.serve(request)
181    /// 3. Internal: let response = jit_layers.layer(http_connection).serve(request)
182    pub fn with_jit_layer<T>(self, jit_layers: T) -> EasyHttpWebClient<BodyIn, ConnResponse, T> {
183        EasyHttpWebClient {
184            connector: self.connector,
185            jit_layers,
186        }
187    }
188}
189
190impl<Body, ConnectionBody, Connection, L> Service<Request<Body>>
191    for EasyHttpWebClient<Body, EstablishedClientConnection<Connection, Request<ConnectionBody>>, L>
192where
193    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
194    Connection:
195        Service<Request<ConnectionBody>, Output = Response, Error = BoxError> + ExtensionsRef,
196    // Body type this connection will be able to send, this is not necessarily the same one that
197    // was used in the request that created this connection
198    ConnectionBody:
199        StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
200    L: Layer<
201            Connection,
202            Service: Service<Request<ConnectionBody>, Output = Response, Error = BoxError>,
203        > + Send
204        + Sync
205        + 'static,
206{
207    type Output = Response;
208    type Error = OpaqueError;
209
210    async fn serve(&self, req: Request<Body>) -> Result<Self::Output, Self::Error> {
211        let uri = req.uri().clone();
212
213        let EstablishedClientConnection {
214            input: req,
215            conn: http_connection,
216        } = self.connector.serve(req).await.into_opaque_error()?;
217
218        req.extensions()
219            .insert(Egress(http_connection.extensions().clone()));
220
221        let http_connection = self.jit_layers.layer(http_connection);
222
223        // NOTE: stack might change request version based on connector data,
224        tracing::trace!(url.full = %uri, "send http req to connector stack");
225
226        let result = http_connection.serve(req).await;
227
228        match result {
229            Ok(resp) => {
230                tracing::trace!(url.full = %uri, "response received from connector stack");
231                Ok(resp)
232            }
233            Err(err) => Err(err
234                .context("http request failure")
235                .context_field("uri", uri)
236                .into_opaque_error()),
237        }
238    }
239}
240
241#[cfg(test)]
242mod tests {
243    use std::{
244        convert::Infallible,
245        sync::{
246            Arc,
247            atomic::{AtomicUsize, Ordering},
248        },
249        time::Duration,
250    };
251
252    use rama_core::service::service_fn;
253    use rama_http::{Body, BodyExtractExt, Version};
254    use rama_http_backend::server::HttpServer;
255    use rama_net::test_utils::client::{MockConnectorService, MockSocket};
256    use serde::{Deserialize, Serialize};
257    use tokio::time::sleep;
258
259    use super::*;
260
261    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
262    struct Output {
263        conn: usize,
264        resp: usize,
265    }
266
267    fn dummy_server<Input: Send + 'static>()
268    -> impl Service<Input, Output = EstablishedClientConnection<MockSocket, Input>, Error = Infallible>
269    {
270        let created_connections = Arc::new(AtomicUsize::new(0));
271        MockConnectorService::new(move || {
272            let created_connections = created_connections.clone();
273            let conn = created_connections.fetch_add(1, Ordering::Relaxed);
274
275            // count responses created on this specific connection
276            let created_response = Arc::new(AtomicUsize::new(0));
277
278            HttpServer::auto(Executor::default()).service(service_fn(move |_req: Request| {
279                let created_response = created_response.clone();
280                let resp = created_response.fetch_add(1, Ordering::Relaxed);
281                async move {
282                    sleep(Duration::from_millis(5)).await;
283                    let out = Output { conn, resp };
284                    let resp = Response::new(Body::from(serde_json::to_vec(&out).unwrap()));
285                    Ok::<_, Infallible>(resp)
286                }
287            }))
288        })
289    }
290
291    #[tokio::test]
292    async fn connection_is_in_use_until_response_body_is_consumed() {
293        let client = EasyHttpWebClient::connector_builder()
294            .with_custom_transport_connector(dummy_server())
295            .without_dns_connector()
296            .without_tls_proxy_support()
297            .without_proxy_support()
298            .without_tls_support()
299            .with_default_http_connector(Executor::default())
300            .try_with_connection_pool(HttpPooledConnectorConfig {
301                max_concurrent_streams: 1,
302                max_total: 4,
303                ..Default::default()
304            })
305            .unwrap()
306            .build_client();
307
308        let req = || {
309            Request::builder()
310                .uri("http://example.com")
311                .version(Version::HTTP_2)
312                .body(Body::empty())
313                .unwrap()
314        };
315
316        // Get the first response but DO NOT consume its body yet: the connection
317        // is logically still in use until the body is drained. Then issue a second
318        // request before draining the first.
319        let res1 = client.serve(req()).await.unwrap();
320        let res2 = client.serve(req()).await.unwrap();
321
322        // Drain in reverse so `res1`'s body is still outstanding when `req2` runs.
323        let out2 = res2.try_into_json::<Output>().await.unwrap();
324        let out1 = res1.try_into_json::<Output>().await.unwrap();
325
326        assert_eq!(out1.conn, 0, "first request uses the first connection");
327        // With `max_concurrent_streams = 1`, connection 0's response body is still
328        // in flight, so the second request must NOT reuse it.
329        assert_eq!(
330            out2.conn, 1,
331            "second request must not reuse a connection whose response body is still in flight"
332        );
333    }
334
335    // These things are already tested inside the pool itself, but here we add some high level tests
336    // in case we ever swap the underlying pool implementation.
337
338    #[tokio::test]
339    async fn default_pool_multiplexes_on_h2() {
340        let client = EasyHttpWebClient::connector_builder()
341            .with_custom_transport_connector(dummy_server())
342            .without_dns_connector()
343            .without_tls_proxy_support()
344            .without_proxy_support()
345            .without_tls_support()
346            .with_default_http_connector(Executor::default())
347            .try_with_default_connection_pool()
348            .unwrap()
349            .build_client();
350
351        let req = || {
352            Request::builder()
353                .uri("http://example.com")
354                .version(Version::HTTP_2)
355                .body(Body::empty())
356                .unwrap()
357        };
358        let (res1, res2, res3) = tokio::join!(
359            client.serve(req()),
360            client.serve(req()),
361            client.serve(req()),
362        );
363
364        // Should only create single connection and send all requests over the same one
365        for (i, res) in [res1, res2, res3].into_iter().enumerate() {
366            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
367            assert_eq!(out.conn, 0);
368            assert_eq!(out.resp, i);
369        }
370    }
371
372    #[tokio::test]
373    async fn default_pool_does_not_multiplexes_on_h1() {
374        let client = EasyHttpWebClient::connector_builder()
375            .with_custom_transport_connector(dummy_server())
376            .without_dns_connector()
377            .without_tls_proxy_support()
378            .without_proxy_support()
379            .without_tls_support()
380            .with_default_http_connector(Executor::default())
381            .try_with_default_connection_pool()
382            .unwrap()
383            .build_client();
384
385        let req = || {
386            Request::builder()
387                .uri("http://example.com")
388                .version(Version::HTTP_11)
389                .body(Body::empty())
390                .unwrap()
391        };
392        let (res1, res2, res3) = tokio::join!(
393            client.serve(req()),
394            client.serve(req()),
395            client.serve(req()),
396        );
397
398        // Should create a new connection for each request since they are all inprogress at the same
399        // time and h1 does not support multiplexing
400        for (i, res) in [res1, res2, res3].into_iter().enumerate() {
401            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
402            assert_eq!(out.conn, i);
403            assert_eq!(out.resp, 0);
404        }
405    }
406
407    #[tokio::test]
408    async fn multiplex_on_h2_respects_limits() {
409        let client = EasyHttpWebClient::connector_builder()
410            .with_custom_transport_connector(dummy_server())
411            .without_dns_connector()
412            .without_tls_proxy_support()
413            .without_proxy_support()
414            .without_tls_support()
415            .with_default_http_connector(Executor::default())
416            .try_with_connection_pool(HttpPooledConnectorConfig {
417                max_concurrent_streams: 2,
418                ..Default::default()
419            })
420            .unwrap()
421            .build_client();
422
423        let req = || {
424            Request::builder()
425                .uri("http://example.com")
426                .version(Version::HTTP_2)
427                .body(Body::empty())
428                .unwrap()
429        };
430        let (res1, res2, res3, res4) = tokio::join!(
431            client.serve(req()),
432            client.serve(req()),
433            client.serve(req()),
434            client.serve(req()),
435        );
436
437        // Should create a connection for every two request
438        for (i, res) in [res1, res2, res3, res4].into_iter().enumerate() {
439            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
440            assert_eq!(out.conn, i / 2);
441            assert_eq!(out.resp, i % 2);
442        }
443    }
444}