simple_hyper_client/
async_client.rs

1/* Copyright (c) Fortanix, Inc.
2 *
3 * This Source Code Form is subject to the terms of the Mozilla Public
4 * License, v. 2.0. If a copy of the MPL was not distributed with this
5 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
6
7use crate::body::RequestBody;
8use crate::connector::{ConnectorAdapter, NetworkConnector};
9use crate::error::Error;
10use crate::{HyperClient, HyperClientBuilder, Response};
11
12use headers::{Header, HeaderMap, HeaderMapExt};
13use hyper::body::Body;
14use hyper::{Method, Request, Uri};
15
16use std::convert::{TryFrom, TryInto};
17use std::fmt;
18use std::future::Future;
19use std::sync::Arc;
20use std::time::Duration;
21
22/// A wrapper for [hyper's `Client` type] providing a simpler interface
23///
24/// Example usage:
25/// ```ignore
26/// let connector = HttpConnector::new();
27/// let client = Client::with_connector(connector);
28/// let response = client.get("http://example.com/")?.send().await?;
29/// ```
30///
31/// [hyper's `Client` type]: https://docs.rs/hyper-util/latest/hyper_util/client/legacy/struct.Client.html
32#[derive(Clone)]
33pub struct Client {
34    inner: Arc<HyperClient<ConnectorAdapter, RequestBody>>,
35}
36
37macro_rules! define_method_fn {
38    (@internal $name:ident, $method:ident, $method_str:expr) => {
39        #[doc = "Initiate a "]
40        #[doc = $method_str]
41        #[doc = " request with the specified URI."]
42        ///
43        /// Returns an error if `uri` is invalid.
44        pub fn $name<U>(&self, uri: U) -> Result<RequestBuilder<'_>, Error>
45        where
46            Uri: TryFrom<U>,
47            <Uri as TryFrom<U>>::Error: Into<http::Error>,
48        {
49            self.request(Method::$method, uri)
50        }
51    };
52
53    ($name:ident, $method:ident) => {
54        define_method_fn!(@internal $name, $method, stringify!($method));
55    };
56}
57
58impl Client {
59    pub fn builder() -> ClientBuilder {
60        ClientBuilder::default()
61    }
62
63    /// Create a new `Client` using the specified connector.
64    pub fn with_connector<C: NetworkConnector>(connector: C) -> Self {
65        ClientBuilder::default().build(connector)
66    }
67
68    /// This method can be used instead of [Client::request]
69    /// if the caller already has a [Request].
70    pub async fn send(&self, request: Request<RequestBody>) -> Result<Response, Error> {
71        Ok(self.inner.request(request).await?)
72    }
73
74    /// Initiate a request with the specified method and URI.
75    ///
76    /// Returns an error if `uri` is invalid.
77    pub fn request<U>(&self, method: Method, uri: U) -> Result<RequestBuilder<'_>, Error>
78    where
79        Uri: TryFrom<U>,
80        <Uri as TryFrom<U>>::Error: Into<http::Error>,
81    {
82        let uri = uri.try_into().map_err(Into::into).map_err(Error::Http)?;
83        Ok(RequestBuilder {
84            client: self,
85            details: RequestDetails::new(method, uri),
86        })
87    }
88
89    define_method_fn!(get, GET);
90    define_method_fn!(head, HEAD);
91    define_method_fn!(post, POST);
92    define_method_fn!(patch, PATCH);
93    define_method_fn!(put, PUT);
94    define_method_fn!(delete, DELETE);
95}
96
97// NOTE: the default values are taken from https://docs.rs/hyper-util/latest/hyper_util/client/legacy/struct.Builder.html
98// NOTE: not all configurable aspects of hyper Client are exposed here.
99/// A builder for [`Client`]
100///
101/// [`Client`]: struct.Client.html
102#[derive(Clone)]
103pub struct ClientBuilder(HyperClientBuilder);
104
105impl Default for ClientBuilder {
106    fn default() -> Self {
107        Self::new()
108    }
109}
110
111impl ClientBuilder {
112    pub fn new() -> Self {
113        Self::from_hyper_client_builder(HyperClientBuilder::new(TokioExecutor))
114    }
115
116    /// Create a builder with a configured instance of [`HyperClientBuilder`].
117    pub fn from_hyper_client_builder(inner: HyperClientBuilder) -> Self {
118        Self(inner)
119    }
120
121    /// Sets the maximum idle connection per host allowed in the pool.
122    ///
123    /// Default is usize::MAX (no limit).
124    pub fn pool_max_idle_per_host(&mut self, max_idle: usize) -> &mut Self {
125        self.0.pool_max_idle_per_host(max_idle);
126        self
127    }
128
129    /// Set an optional timeout for idle sockets being kept-alive.
130    ///
131    /// Pass `None` to disable timeout.
132    ///
133    /// Default is 90 seconds.
134    pub fn pool_idle_timeout(&mut self, val: Option<Duration>) -> &mut Self {
135        self.0.pool_idle_timeout(val);
136        self
137    }
138
139    /// Set whether the connection **must** use HTTP/2.
140    ///
141    /// Note that setting this to true prevents HTTP/1 from being allowed.
142    ///
143    /// Default is false.
144    pub fn http2_only(&mut self, val: bool) -> &mut Self {
145        self.0.http2_only(val);
146        self
147    }
148
149    /// Combine the configuration of this builder with a connector to create a
150    /// `Client`.
151    pub fn build<C: NetworkConnector>(&self, connector: C) -> Client {
152        Client {
153            inner: Arc::new(self.0.build(ConnectorAdapter::new(connector))),
154        }
155    }
156}
157
158pub(crate) struct RequestDetails {
159    pub(crate) method: Method,
160    pub(crate) uri: Uri,
161    pub(crate) headers: HeaderMap,
162    pub(crate) body: Option<RequestBody>,
163}
164
165impl fmt::Debug for RequestDetails {
166    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
167        f.debug_struct("RequestDetails")
168            .field("method", &self.method)
169            .field("uri", &self.uri)
170            .field("headers", &self.headers.len())
171            .field("body", &self.body.as_ref().map_or("None", |_| "Some(...)"))
172            .finish()
173    }
174}
175
176impl RequestDetails {
177    pub fn new(method: Method, uri: Uri) -> Self {
178        RequestDetails {
179            method,
180            uri,
181            headers: HeaderMap::new(),
182            body: None,
183        }
184    }
185
186    pub async fn send(self, client: &Client) -> Result<Response, Error> {
187        let req = self.into_request()?;
188        Ok(client.inner.request(req).await?)
189    }
190
191    pub fn into_request(self) -> Result<Request<RequestBody>, Error> {
192        let can_have_body = match self.method {
193            // See RFC 7231 section 4.3
194            Method::GET | Method::HEAD | Method::DELETE => false,
195            _ => true,
196        };
197        let body = if can_have_body {
198            self.body.unwrap_or_else(|| RequestBody::empty())
199        } else if self.body.is_some_and(|body| body.size_hint().lower() > 0) {
200            return Err(Error::BodyNotAllowed(self.method));
201        } else {
202            RequestBody::empty()
203        };
204        let mut req = Request::builder().method(self.method).uri(self.uri);
205        match req.headers_mut() {
206            Some(headers) => {
207                *headers = self.headers;
208            }
209            // There is an error in req, but the only way to extract the error is through `req.body()`
210            None => match req.body(RequestBody::empty()) {
211                Err(e) => return Err(e.into()),
212                Ok(_) => {
213                    panic!("request builder must have errors if `fn headers_mut()` returns None")
214                }
215            },
216        }
217
218        Ok(req.body(body)?)
219    }
220}
221
222/// An HTTP request builder
223///
224/// This is created through [`Client::get()`], [`Client::post()`] etc.
225/// You need to call [`send()`] to actually send the request over the network.
226/// If you don't want to send it and just want the resultant [Request], you
227/// can call [RequestBuilder::build].
228///
229/// [`Client::get()`]: struct.Client.html#method.get
230/// [`Client::post()`]: struct.Client.html#method.post
231/// [`send()`]: struct.RequestBuilder.html#method.send
232pub struct RequestBuilder<'a> {
233    client: &'a Client,
234    details: RequestDetails,
235}
236
237impl<'a> RequestBuilder<'a> {
238    /// Set the request body.
239    pub fn body<B: Into<RequestBody>>(mut self, body: B) -> Self {
240        self.details.body = Some(body.into());
241        self
242    }
243
244    /// Set the request headers.
245    pub fn headers(mut self, headers: HeaderMap) -> Self {
246        self.details.headers = headers;
247        self
248    }
249
250    /// Set a single header using [`HeaderMapExt::typed_insert()`].
251    ///
252    /// [`HeaderMapExt::typed_insert()`]: https://docs.rs/headers/0.3.5/headers/trait.HeaderMapExt.html#tymethod.typed_insert
253    pub fn header<H: Header>(mut self, header: H) -> Self {
254        self.details.headers.typed_insert(header);
255        self
256    }
257
258    /// Get the resultant [Request].
259    ///
260    /// Prefer [RequestBuilder::send] unless you have a specific
261    /// need to get the resultant [Request].
262    pub fn build(self) -> Result<Request<RequestBody>, Error> {
263        self.details.into_request()
264    }
265
266    /// Send the request over the network.
267    ///
268    /// Returns an error before sending the request if there is something wrong
269    /// with the request parameters (method, uri, etc.).
270    pub async fn send(self) -> Result<Response, Error> {
271        self.details.send(&self.client).await
272    }
273}
274
275/// The default executor to use with the native hyper client.
276///
277/// Based on the [`tokio`] runtime.
278#[derive(Copy, Clone)]
279pub struct TokioExecutor;
280
281impl<F> hyper::rt::Executor<F> for TokioExecutor
282where
283    F: Future + Send + 'static,
284    F::Output: Send + 'static,
285{
286    fn execute(&self, fut: F) {
287        tokio::spawn(fut);
288    }
289}
290
291#[cfg(test)]
292mod tests {
293    use super::*;
294    use crate::connector::HttpConnector;
295    use crate::util::to_bytes;
296    use crate::SharedBody;
297
298    use headers::{ContentLength, ContentType};
299    use hyper::body::Bytes;
300    use hyper::StatusCode;
301    use test_case::test_case;
302    use tokio::io::{AsyncReadExt, AsyncWriteExt};
303    use tokio::net::TcpListener;
304    use tokio::sync::oneshot;
305
306    use std::net::SocketAddr;
307
308    const RESPONSE_OK: &str = "HTTP/1.1 200 OK\r\nContent-Length: 13\r\n\r\nHello, world!\r\n";
309    const RESPONSE_404: &str =
310        "HTTP/1.1 404 Not Found\r\nContent-Length: 23\r\n\r\nResource was not found.\r\n";
311
312    async fn test_http_server(resp: &'static str, body_tx: oneshot::Sender<Vec<u8>>) -> SocketAddr {
313        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
314        let addr = listener.local_addr().unwrap();
315        tokio::spawn(async move {
316            let (mut stream, _) = listener.accept().await.unwrap();
317            let mut input = Vec::new();
318            stream.write_all(resp.as_bytes()).await.unwrap();
319            stream.read_to_end(&mut input).await.unwrap();
320            let _ = body_tx.send(input);
321        });
322        addr
323    }
324
325    #[test_case(r#"{"key":"value"}"#; "with body")]
326    #[test_case(SharedBody::empty(); "without body")]
327    #[tokio::test]
328    async fn http_client<B: Into<SharedBody>>(req_body: B) {
329        let (tx, rx) = oneshot::channel();
330        let addr = test_http_server(RESPONSE_OK, tx).await;
331        let url = format!("http://{}/", addr);
332
333        let req_body = req_body.into();
334        let expected_content_length = req_body.size_hint().exact();
335
336        let connector = HttpConnector::new();
337        let client = Client::with_connector(connector);
338        let response = client
339            .post(url)
340            .unwrap()
341            .header(ContentType::json())
342            .body(req_body.clone())
343            .send()
344            .await
345            .unwrap();
346
347        // Parse the request received by the server
348        let mut headers = [httparse::EMPTY_HEADER; 64];
349        let mut request = httparse::Request::new(&mut headers);
350        let req_buf = rx.await.unwrap();
351        let body_idx = request.parse(&req_buf).unwrap().unwrap();
352        assert_eq!(request.method, Some("POST"));
353        assert_eq!(request.path, Some("/"));
354        assert_eq!(request.version, Some(1));
355        let content_length = request
356            .headers
357            .iter()
358            .find(|header| header.name == ContentLength::name());
359        if let Some(expected_content_length) = expected_content_length {
360            assert_eq!(
361                content_length.unwrap().value,
362                expected_content_length.to_string().as_bytes()
363            );
364        } else {
365            assert_eq!(content_length, None);
366        }
367        assert_eq!(&req_buf[body_idx..], req_body.as_ref());
368
369        assert_eq!(response.status(), StatusCode::OK);
370        let response_body = to_bytes(response).await.unwrap();
371        assert_eq!(response_body, "Hello, world!".as_bytes());
372    }
373
374    #[test_case(Some(r#"{"key":"value"}"#.into()), false; "non-empty body not allowed")]
375    #[test_case(Some("".into()), true; "empty body allowed")]
376    #[test_case(None, true; "without body allowed")]
377    #[tokio::test]
378    async fn get_request(body: Option<RequestBody>, expect_ok: bool) {
379        let (tx, _rx) = oneshot::channel();
380        let addr = test_http_server(RESPONSE_OK, tx).await;
381        let url = format!("http://{}/", addr);
382
383        let connector = HttpConnector::new();
384        let client = Client::with_connector(connector);
385        let mut builder = client.get(url).unwrap();
386
387        if let Some(body) = body {
388            builder = builder.header(ContentType::json()).body(body);
389        }
390
391        let result = builder.send().await;
392        if expect_ok {
393            result.unwrap();
394        } else {
395            assert_eq!(result.unwrap_err().unwrap_body_not_allowed(), Method::GET);
396        }
397    }
398
399    #[tokio::test]
400    async fn drop_client_before_response() {
401        let (tx, _rx) = oneshot::channel();
402        let addr = test_http_server(RESPONSE_404, tx).await;
403        let url = format!("http://{}/", addr);
404
405        let connector = HttpConnector::new();
406        let client = Client::with_connector(connector);
407        let response = client.get(url).unwrap().send().await.unwrap();
408        drop(client);
409
410        assert_eq!(response.status(), StatusCode::NOT_FOUND);
411        assert_eq!(response.headers().len(), 1);
412        let body = to_bytes(response).await.unwrap();
413        assert_eq!(body, "Resource was not found.");
414    }
415
416    #[tokio::test]
417    async fn http_connector_connect_timeout() {
418        // IP address chosen from 192.0.2.0/24 block defined in RFC 5737.
419        let url = "http://192.0.2.1/";
420        let connector = HttpConnector::new().connect_timeout(Some(Duration::from_millis(100)));
421        let client = Client::with_connector(connector);
422        let err = client.get(url).unwrap().send().await.unwrap_err();
423        assert_eq!(
424            err.to_string(),
425            "client error (Connect): I/O error: connection timed out"
426        );
427    }
428
429    #[test]
430    fn wrapped_body_size_hint() {
431        let body = http_body_util::Full::new(Bytes::from_static(r#"{"key":"value"}"#.as_bytes()));
432        let unwrapped_size_hint = body.size_hint();
433
434        let request = Client::with_connector(HttpConnector::new())
435            .post("https://localhost/")
436            .unwrap()
437            .header(ContentType::json())
438            .body(RequestBody::wrap(body))
439            .build()
440            .unwrap();
441
442        assert_eq!(
443            request.body().size_hint().lower(),
444            unwrapped_size_hint.lower()
445        );
446        assert_eq!(
447            request.body().size_hint().upper(),
448            unwrapped_size_hint.upper()
449        );
450    }
451}
simple_hyper_client/async_client.rs

simple_hyper_client/
async_client.rs
