simple_hyper_client/blocking/
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 super::body::Body;
8use super::Response;
9use crate::async_client::{ClientBuilder as AsyncClientBuilder, RequestDetails};
10use crate::body::RequestBody;
11use crate::connector::NetworkConnector;
12use crate::error::Error;
13
14use futures_executor::block_on;
15use headers::{Header, HeaderMap, HeaderMapExt};
16use hyper::{Method, Uri};
17use hyper_util::client::legacy::Builder as HyperClientBuilder;
18use tokio::runtime;
19use tokio::sync::{mpsc, oneshot};
20
21use std::convert::{TryFrom, TryInto};
22use std::sync::Arc;
23use std::thread::{self, JoinHandle};
24use std::time::Duration;
25
26/// A wrapper for [hyper's `Client` type] providing a blocking interface
27///
28/// Example usage:
29/// ```ignore
30/// let connector = HttpConnector::new();
31/// let client = Client::with_connector(connector);
32/// let response = client.get("http://example.com/")?.send()?;
33/// ```
34///
35/// [hyper's `Client` type]: https://docs.rs/hyper-util/latest/hyper_util/client/legacy/struct.Client.html
36#[derive(Clone)]
37pub struct Client {
38    inner: Arc<ClientInner>,
39}
40
41type ResponseSender = oneshot::Sender<Result<Response, Error>>;
42
43struct ClientInner {
44    tx: Option<mpsc::UnboundedSender<(RequestDetails, ResponseSender)>>,
45    thread: Option<JoinHandle<()>>,
46}
47
48impl Drop for ClientInner {
49    fn drop(&mut self) {
50        // signal shutdown to the thread
51        self.tx.take();
52        self.thread.take().map(|h| h.join());
53    }
54}
55
56macro_rules! define_method_fn {
57    (@internal $name:ident, $method:ident, $method_str:expr) => {
58        #[doc = "Initiate a "]
59        #[doc = $method_str]
60        #[doc = " request with the specified URI."]
61        ///
62        /// Returns an error if `uri` is invalid.
63        pub fn $name<U>(&self, uri: U) -> Result<RequestBuilder<'_>, Error>
64        where
65            Uri: TryFrom<U>,
66            <Uri as TryFrom<U>>::Error: Into<http::Error>,
67        {
68            self.request(Method::$method, uri)
69        }
70    };
71
72    ($name:ident, $method:ident) => {
73        define_method_fn!(@internal $name, $method, stringify!($method));
74    };
75}
76
77impl Client {
78    pub fn builder() -> ClientBuilder {
79        ClientBuilder::default()
80    }
81
82    pub fn with_connector<C: NetworkConnector>(connector: C) -> Self {
83        ClientBuilder::default().build(connector)
84    }
85
86    /// Initiate a request with the specified method and URI.
87    ///
88    /// Returns an error if `uri` is invalid.
89    pub fn request<U>(&self, method: Method, uri: U) -> Result<RequestBuilder<'_>, Error>
90    where
91        Uri: TryFrom<U>,
92        <Uri as TryFrom<U>>::Error: Into<http::Error>,
93    {
94        let uri = uri.try_into().map_err(Into::into).map_err(Error::Http)?;
95        Ok(RequestBuilder {
96            client: self,
97            details: RequestDetails::new(method, uri),
98        })
99    }
100
101    define_method_fn!(get, GET);
102    define_method_fn!(head, HEAD);
103    define_method_fn!(post, POST);
104    define_method_fn!(patch, PATCH);
105    define_method_fn!(put, PUT);
106    define_method_fn!(delete, DELETE);
107}
108
109/// A builder for [`Client`].
110///
111/// [`Client`]: struct.Client.html
112#[derive(Default, Clone)]
113pub struct ClientBuilder(AsyncClientBuilder);
114
115impl ClientBuilder {
116    pub fn new() -> Self {
117        Self::default()
118    }
119
120    /// Create a builder with a configured instance of [`HyperClientBuilder`].
121    pub fn from_hyper_client_builder(inner: HyperClientBuilder) -> Self {
122        Self(AsyncClientBuilder::from_hyper_client_builder(inner))
123    }
124
125    /// Sets the maximum idle connection per host allowed in the pool.
126    ///
127    /// Default is usize::MAX (no limit).
128    pub fn pool_max_idle_per_host(&mut self, max_idle: usize) -> &mut Self {
129        self.0.pool_max_idle_per_host(max_idle);
130        self
131    }
132
133    /// Set an optional timeout for idle sockets being kept-alive.
134    ///
135    /// Pass `None` to disable timeout.
136    ///
137    /// Default is 90 seconds.
138    pub fn pool_idle_timeout(&mut self, val: Option<Duration>) -> &mut Self {
139        self.0.pool_idle_timeout(val);
140        self
141    }
142
143    /// Set whether the connection **must** use HTTP/2.
144    ///
145    /// Note that setting this to true prevents HTTP/1 from being allowed.
146    ///
147    /// Default is false.
148    pub fn http2_only(&mut self, val: bool) -> &mut Self {
149        self.0.http2_only(val);
150        self
151    }
152
153    /// Combine the configuration of this builder with a connector to create a
154    /// `Client`.
155    pub fn build<C: NetworkConnector>(&self, connector: C) -> Client {
156        let async_client = self.0.build(connector);
157        let (tx, mut rx) = mpsc::unbounded_channel::<(RequestDetails, ResponseSender)>();
158
159        let thread = thread::spawn(move || {
160            let rt = runtime::Builder::new_current_thread()
161                .enable_all()
162                .build()
163                .unwrap(); // TODO: send back an error through a oneshot channel
164
165            rt.block_on(async move {
166                while let Some((req_details, resp_tx)) = rx.recv().await {
167                    let async_client = async_client.clone();
168                    tokio::spawn(async move {
169                        match req_details.send(&async_client).await {
170                            Ok(resp) => {
171                                let (parts, hyper_body) = resp.into_parts();
172                                let (fut, body) = Body::new(hyper_body);
173                                let _ = resp_tx.send(Ok(Response::from_parts(parts, body)));
174                                fut.await;
175                            }
176                            Err(e) => {
177                                let _: Result<_, _> = resp_tx.send(Err(e));
178                            }
179                        }
180                    });
181                }
182            })
183        });
184
185        Client {
186            inner: Arc::new(ClientInner {
187                tx: Some(tx),
188                thread: Some(thread),
189            }),
190        }
191    }
192}
193
194/// An HTTP request builder
195///
196/// This is created through [`Client::get()`], [`Client::post()`] etc.
197/// You need to call [`send()`] to actually send the request over the network.
198///
199/// [`Client::get()`]: struct.Client.html#method.get
200/// [`Client::post()`]: struct.Client.html#method.post
201/// [`send()`]: struct.RequestBuilder.html#method.send
202pub struct RequestBuilder<'a> {
203    client: &'a Client,
204    details: RequestDetails,
205}
206
207impl<'a> RequestBuilder<'a> {
208    /// Set the request body.
209    pub fn body<B: Into<RequestBody>>(mut self, body: B) -> Self {
210        self.details.body = Some(body.into());
211        self
212    }
213
214    /// Set the request headers.
215    pub fn headers(mut self, headers: HeaderMap) -> Self {
216        self.details.headers = headers;
217        self
218    }
219
220    /// Set a single header using [`HeaderMapExt::typed_insert()`].
221    ///
222    /// [`HeaderMapExt::typed_insert()`]: https://docs.rs/headers/0.3.5/headers/trait.HeaderMapExt.html#tymethod.typed_insert
223    pub fn header<H: Header>(mut self, header: H) -> Self {
224        self.details.headers.typed_insert(header);
225        self
226    }
227
228    /// Send the request over the network.
229    ///
230    /// Returns an error before sending the request if there is something wrong
231    /// with the request parameters (method, uri, etc.).
232    pub fn send(self) -> Result<Response, Error> {
233        let RequestBuilder { client, details } = self;
234        let (tx, rx) = oneshot::channel();
235        client
236            .inner
237            .tx
238            .as_ref()
239            .expect("runtime thread exited early")
240            .send((details, tx))
241            .expect("runtime thread panicked");
242
243        // TODO: replace `block_on` with `rx.blocking_recv()` once we move to tokio 1.16+
244        block_on(async move {
245            match rx.await {
246                Ok(res) => res,
247                Err(_) => panic!("event loop panicked"),
248            }
249        })
250        .map(|mut resp| {
251            resp.body_mut().keep_client_alive = KeepClientAlive(Some(client.inner.clone()));
252            resp
253        })
254    }
255}
256
257pub(super) struct KeepClientAlive(#[allow(unused)] Option<Arc<ClientInner>>);
258
259impl KeepClientAlive {
260    pub fn empty() -> Self {
261        KeepClientAlive(None)
262    }
263}
264
265#[cfg(test)]
266mod tests {
267    use super::*;
268    use crate::connector::HttpConnector;
269    use headers::ContentType;
270    use hyper::StatusCode;
271    use std::io::{Read, Write};
272    use std::net::{SocketAddr, TcpListener};
273    use std::thread;
274
275    const RESPONSE_OK: &str = "HTTP/1.1 200 OK\r\nContent-Length: 13\r\n\r\nHello, world!\r\n";
276    const RESPONSE_404: &str =
277        "HTTP/1.1 404 Not Found\r\nContent-Length: 23\r\n\r\nResource was not found.\r\n";
278
279    fn test_http_server(resp: &'static str) -> SocketAddr {
280        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
281        let addr = listener.local_addr().unwrap();
282        thread::spawn(move || {
283            let (mut stream, _) = listener.accept().unwrap();
284            let mut input = Vec::new();
285            stream.read(&mut input).unwrap();
286            stream.write_all(resp.as_bytes()).unwrap();
287        });
288        addr
289    }
290
291    #[test]
292    fn http_client_ok() {
293        let addr = test_http_server(RESPONSE_OK);
294        let url = format!("http://{}/", addr);
295
296        let connector = HttpConnector::new();
297        let client = Client::with_connector(connector);
298        let mut response = client
299            .request(Method::POST, url)
300            .unwrap()
301            .header(ContentType::json())
302            .body(r#"{"key":"value"}"#)
303            .send()
304            .unwrap();
305
306        assert_eq!(response.status(), StatusCode::OK);
307        let mut body = String::new();
308        response.body_mut().read_to_string(&mut body).unwrap();
309        assert_eq!(body, "Hello, world!");
310    }
311
312    #[test]
313    fn drop_client_before_response() {
314        let addr = test_http_server(RESPONSE_404);
315        let url = format!("http://{}/", addr);
316
317        let connector = HttpConnector::new();
318        let client = Client::with_connector(connector);
319        let mut response = client.get(url).unwrap().send().unwrap();
320        drop(client);
321
322        assert_eq!(response.status(), StatusCode::NOT_FOUND);
323        assert_eq!(response.headers().len(), 1);
324        let mut body = String::new();
325        response.body_mut().read_to_string(&mut body).unwrap();
326        assert_eq!(body, "Resource was not found.");
327    }
328}
simple_hyper_client/blocking/client.rs

simple_hyper_client/blocking/
client.rs
