rustycrawl_http_client/
lib.rs

1//! A small reqwest-like HTTP client built over a caller-provided Hyper service.
2//!
3//! The client automatically stores `Set-Cookie` response headers and attaches
4//! matching cookies to later requests. The cookie store is enabled by default.
5
6use std::{
7    convert::Infallible,
8    error::Error as StdError,
9    future::Future,
10    sync::{Arc, RwLock},
11    time::Duration,
12};
13
14use cookie::Cookie;
15use cookie_store::CookieStore;
16use http::{HeaderMap, HeaderName, HeaderValue, Method, Request, Uri, header};
17use http_body_util::{BodyExt, Full, combinators::BoxBody};
18use hyper::{Response as HyperResponse, body::Incoming};
19use hyper_rustls::HttpsConnector;
20use hyper_util::{
21    client::legacy::{self, connect::HttpConnector},
22    rt::TokioExecutor,
23};
24use serde::{Serialize, de::DeserializeOwned};
25use url::Url;
26
27pub use http::{StatusCode, Version};
28pub use hyper::body::Bytes;
29
30pub type BoxError = Box<dyn StdError + Send + Sync>;
31
32/// The conventional Tokio connector used for direct HTTP and HTTPS requests.
33pub type StandardConnector = HttpsConnector<HttpConnector>;
34
35/// A raw Hyper client using [`StandardConnector`].
36pub type StandardHyperClient = legacy::Client<StandardConnector, Body>;
37
38/// A Hyper transport service using the regular Tokio networking stack.
39#[derive(Clone)]
40pub struct StandardService(StandardHyperClient);
41
42impl StandardService {
43    /// Access the underlying raw Hyper client.
44    #[must_use]
45    pub const fn hyper_client(&self) -> &StandardHyperClient {
46        &self.0
47    }
48}
49
50impl HttpService for StandardService {
51    type Error = legacy::Error;
52
53    async fn send(&self, request: Request<Body>) -> Result<HyperResponse<Incoming>, Self::Error> {
54        self.0.request(request).await
55    }
56}
57
58/// A cookie-aware client using the regular Tokio networking stack.
59pub type StandardClient = Client<StandardService>;
60
61/// A reqwest-like builder for a standard client, including Hyper pool tuning.
62pub struct StandardClientBuilder {
63    cookie_store: bool,
64    default_headers: HeaderMap,
65    timeout: Option<Duration>,
66    pool_idle_timeout: Option<Option<Duration>>,
67    pool_max_idle_per_host: Option<usize>,
68}
69
70impl StandardClientBuilder {
71    /// Enable or disable the automatic cookie store.
72    #[must_use]
73    pub fn cookie_store(mut self, enabled: bool) -> Self {
74        self.cookie_store = enabled;
75        self
76    }
77
78    /// Set headers added to requests that do not already contain them.
79    #[must_use]
80    pub fn default_headers(mut self, headers: HeaderMap) -> Self {
81        self.default_headers = headers;
82        self
83    }
84
85    /// Set the timeout for an entire request.
86    #[must_use]
87    pub fn timeout(mut self, timeout: Duration) -> Self {
88        self.timeout = Some(timeout);
89        self
90    }
91
92    /// Set how long idle pooled connections remain available for reuse.
93    ///
94    /// Pass `None` to disable the idle timeout.
95    #[must_use]
96    pub fn pool_idle_timeout<D>(mut self, timeout: D) -> Self
97    where
98        D: Into<Option<Duration>>,
99    {
100        self.pool_idle_timeout = Some(timeout.into());
101        self
102    }
103
104    /// Set the maximum number of idle connections retained per host.
105    #[must_use]
106    pub fn pool_max_idle_per_host(mut self, max: usize) -> Self {
107        self.pool_max_idle_per_host = Some(max);
108        self
109    }
110
111    /// Build the client and its shared Hyper connection pool.
112    #[must_use]
113    pub fn build(self) -> StandardClient {
114        let mut hyper_builder = legacy::Client::builder(TokioExecutor::new());
115        if let Some(timeout) = self.pool_idle_timeout {
116            hyper_builder.pool_idle_timeout(timeout);
117        }
118        if let Some(max) = self.pool_max_idle_per_host {
119            hyper_builder.pool_max_idle_per_host(max);
120        }
121        ClientBuilder {
122            service: StandardService(hyper_builder.build(standard_connector())),
123            cookie_store: self.cookie_store,
124            default_headers: self.default_headers,
125            timeout: self.timeout,
126        }
127        .build()
128    }
129}
130
131/// Build a standard connector supporting HTTP/1.1, HTTP/2, and Web PKI roots.
132#[must_use]
133pub fn standard_connector() -> StandardConnector {
134    // Dependency feature unification can make multiple rustls providers
135    // available in a larger application. Select the same provider used by the
136    // WireGuard connector rather than relying on feature auto-detection.
137    let _ = rustls::crypto::ring::default_provider().install_default();
138    hyper_rustls::HttpsConnectorBuilder::new()
139        .with_webpki_roots()
140        .https_or_http()
141        .enable_http1()
142        .enable_http2()
143        .build()
144}
145
146/// Build a raw Hyper client running on Tokio's networking stack.
147#[must_use]
148pub fn standard_hyper_client() -> StandardHyperClient {
149    legacy::Client::builder(TokioExecutor::new()).build(standard_connector())
150}
151
152/// Build a standard Hyper transport for use with [`Client::builder`].
153#[must_use]
154pub fn standard_service() -> StandardService {
155    StandardService(standard_hyper_client())
156}
157
158/// Build a reqwest-like client with standard networking and cookies enabled.
159#[must_use]
160pub fn standard_client() -> StandardClient {
161    standard_client_builder().build()
162}
163
164/// Build a configurable standard client.
165///
166/// Hyper pools and reuses connections by default. Use this builder to tune the
167/// pool as well as the higher-level cookie, header, and request settings.
168#[must_use]
169pub fn standard_client_builder() -> StandardClientBuilder {
170    StandardClientBuilder {
171        cookie_store: true,
172        default_headers: HeaderMap::new(),
173        timeout: None,
174        pool_idle_timeout: None,
175        pool_max_idle_per_host: None,
176    }
177}
178
179/// The request body accepted by [`HttpService`].
180pub struct Body(BoxBody<Bytes, BoxError>);
181
182impl Body {
183    #[must_use]
184    pub fn empty() -> Self {
185        Self::from(Bytes::new())
186    }
187}
188
189impl Default for Body {
190    fn default() -> Self {
191        Self::empty()
192    }
193}
194
195impl From<Bytes> for Body {
196    fn from(value: Bytes) -> Self {
197        Self(
198            Full::new(value)
199                .map_err(|error: Infallible| match error {})
200                .boxed(),
201        )
202    }
203}
204
205impl From<Vec<u8>> for Body {
206    fn from(value: Vec<u8>) -> Self {
207        Self::from(Bytes::from(value))
208    }
209}
210
211impl From<String> for Body {
212    fn from(value: String) -> Self {
213        Self::from(Bytes::from(value))
214    }
215}
216
217impl From<&'static str> for Body {
218    fn from(value: &'static str) -> Self {
219        Self::from(Bytes::from_static(value.as_bytes()))
220    }
221}
222
223impl hyper::body::Body for Body {
224    type Data = Bytes;
225    type Error = BoxError;
226
227    fn poll_frame(
228        mut self: std::pin::Pin<&mut Self>,
229        cx: &mut std::task::Context<'_>,
230    ) -> std::task::Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
231        std::pin::Pin::new(&mut self.0).poll_frame(cx)
232    }
233
234    fn is_end_stream(&self) -> bool {
235        self.0.is_end_stream()
236    }
237
238    fn size_hint(&self) -> hyper::body::SizeHint {
239        self.0.size_hint()
240    }
241}
242
243/// A cloneable Hyper-compatible request service.
244pub trait HttpService: Clone + Send + Sync + 'static {
245    type Error: StdError + Send + Sync + 'static;
246
247    fn send(
248        &self,
249        request: Request<Body>,
250    ) -> impl Future<Output = Result<HyperResponse<Incoming>, Self::Error>> + Send;
251}
252
253impl<S> HttpService for S
254where
255    S: hyper::service::Service<Request<Body>, Response = HyperResponse<Incoming>>
256        + Clone
257        + Send
258        + Sync
259        + 'static,
260    S::Error: StdError + Send + Sync + 'static,
261    S::Future: Send,
262{
263    type Error = S::Error;
264
265    fn send(
266        &self,
267        request: Request<Body>,
268    ) -> impl Future<Output = Result<HyperResponse<Incoming>, Self::Error>> + Send {
269        self.call(request)
270    }
271}
272
273#[derive(Debug, thiserror::Error)]
274#[non_exhaustive]
275pub enum Error {
276    #[error("invalid URL: {0}")]
277    Url(#[from] url::ParseError),
278    #[error("invalid request: {0}")]
279    Request(#[from] http::Error),
280    #[error("invalid header: {0}")]
281    Header(String),
282    #[error("HTTP transport failed")]
283    Transport(#[source] BoxError),
284    #[error("request timed out")]
285    Timeout,
286    #[error("cookie store lock was poisoned")]
287    CookieStore,
288    #[error("response body failed")]
289    Body(#[source] BoxError),
290    #[error("JSON serialization failed: {0}")]
291    Json(#[from] serde_json::Error),
292    #[error("response was not valid UTF-8: {0}")]
293    Utf8(#[from] std::string::FromUtf8Error),
294}
295
296/// A configurable HTTP client backed by an [`HttpService`].
297#[derive(Clone)]
298pub struct Client<S> {
299    service: S,
300    cookies: Option<Arc<RwLock<CookieStore>>>,
301    default_headers: HeaderMap,
302    timeout: Option<Duration>,
303}
304
305impl<S: HttpService> Client<S> {
306    #[must_use]
307    pub fn new(service: S) -> Self {
308        Self::builder(service).build()
309    }
310
311    #[must_use]
312    pub fn builder(service: S) -> ClientBuilder<S> {
313        ClientBuilder {
314            service,
315            cookie_store: true,
316            default_headers: HeaderMap::new(),
317            timeout: None,
318        }
319    }
320
321    pub fn request<U>(&self, method: Method, url: U) -> Result<RequestBuilder<S>, Error>
322    where
323        U: AsRef<str>,
324    {
325        let url = Url::parse(url.as_ref())?;
326        Ok(RequestBuilder {
327            client: self.clone(),
328            method,
329            url,
330            headers: HeaderMap::new(),
331            body: Body::empty(),
332        })
333    }
334
335    pub fn get<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
336        self.request(Method::GET, url)
337    }
338
339    pub fn head<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
340        self.request(Method::HEAD, url)
341    }
342
343    pub fn post<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
344        self.request(Method::POST, url)
345    }
346
347    pub fn put<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
348        self.request(Method::PUT, url)
349    }
350
351    pub fn patch<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
352        self.request(Method::PATCH, url)
353    }
354
355    pub fn delete<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
356        self.request(Method::DELETE, url)
357    }
358
359    /// Execute an already-built request, applying default headers and cookies.
360    pub async fn execute(&self, request: Request<Body>) -> Result<Response, Error> {
361        let url = Url::parse(&request.uri().to_string())?;
362        self.execute_url(url, request).await
363    }
364
365    async fn execute_url(&self, url: Url, mut request: Request<Body>) -> Result<Response, Error> {
366        for (name, value) in &self.default_headers {
367            if !request.headers().contains_key(name) {
368                request.headers_mut().insert(name, value.clone());
369            }
370        }
371        if !request.headers().contains_key(header::COOKIE) {
372            if let Some(store) = &self.cookies {
373                let store = store.read().map_err(|_| Error::CookieStore)?;
374                if let Some(value) = cookie_header(&store, &url) {
375                    request.headers_mut().insert(
376                        header::COOKIE,
377                        HeaderValue::from_str(&value)
378                            .map_err(|error| Error::Header(error.to_string()))?,
379                    );
380                }
381            }
382        }
383        let sent = self.service.send(request);
384        let response = if let Some(timeout) = self.timeout {
385            tokio::time::timeout(timeout, sent)
386                .await
387                .map_err(|_| Error::Timeout)?
388                .map_err(|error| Error::Transport(Box::new(error)))?
389        } else {
390            sent.await
391                .map_err(|error| Error::Transport(Box::new(error)))?
392        };
393        if let Some(store) = &self.cookies {
394            let mut store = store.write().map_err(|_| Error::CookieStore)?;
395            store_response_cookies(&mut store, response.headers(), &url);
396        }
397        Ok(Response(response))
398    }
399}
400
401fn cookie_header(store: &CookieStore, url: &Url) -> Option<String> {
402    let value = store
403        .get_request_values(url)
404        .map(|(name, value)| format!("{name}={value}"))
405        .collect::<Vec<_>>()
406        .join("; ");
407    (!value.is_empty()).then_some(value)
408}
409
410fn store_response_cookies(store: &mut CookieStore, headers: &HeaderMap, url: &Url) {
411    let parsed = headers
412        .get_all(header::SET_COOKIE)
413        .iter()
414        .filter_map(|value| value.to_str().ok())
415        .filter_map(|value| Cookie::parse(value.to_owned()).ok())
416        .map(Cookie::into_owned);
417    store.store_response_cookies(parsed, url);
418}
419
420pub struct ClientBuilder<S> {
421    service: S,
422    cookie_store: bool,
423    default_headers: HeaderMap,
424    timeout: Option<Duration>,
425}
426
427impl<S> ClientBuilder<S> {
428    #[must_use]
429    pub fn cookie_store(mut self, enabled: bool) -> Self {
430        self.cookie_store = enabled;
431        self
432    }
433
434    #[must_use]
435    pub fn default_headers(mut self, headers: HeaderMap) -> Self {
436        self.default_headers = headers;
437        self
438    }
439
440    #[must_use]
441    pub fn timeout(mut self, timeout: Duration) -> Self {
442        self.timeout = Some(timeout);
443        self
444    }
445
446    #[must_use]
447    pub fn build(self) -> Client<S> {
448        Client {
449            service: self.service,
450            cookies: self
451                .cookie_store
452                .then(|| Arc::new(RwLock::new(CookieStore::default()))),
453            default_headers: self.default_headers,
454            timeout: self.timeout,
455        }
456    }
457}
458
459pub struct RequestBuilder<S> {
460    client: Client<S>,
461    method: Method,
462    url: Url,
463    headers: HeaderMap,
464    body: Body,
465}
466
467impl<S: HttpService> RequestBuilder<S> {
468    #[must_use]
469    pub fn body(mut self, body: impl Into<Body>) -> Self {
470        self.body = body.into();
471        self
472    }
473
474    pub fn header<K, V>(mut self, key: K, value: V) -> Result<Self, Error>
475    where
476        HeaderName: TryFrom<K>,
477        <HeaderName as TryFrom<K>>::Error: std::fmt::Display,
478        HeaderValue: TryFrom<V>,
479        <HeaderValue as TryFrom<V>>::Error: std::fmt::Display,
480    {
481        let key = HeaderName::try_from(key).map_err(|e| Error::Header(e.to_string()))?;
482        let value = HeaderValue::try_from(value).map_err(|e| Error::Header(e.to_string()))?;
483        self.headers.insert(key, value);
484        Ok(self)
485    }
486
487    #[must_use]
488    pub fn headers(mut self, headers: HeaderMap) -> Self {
489        self.headers.extend(headers);
490        self
491    }
492
493    pub fn json(mut self, value: &impl Serialize) -> Result<Self, Error> {
494        self.body = Body::from(serde_json::to_vec(value)?);
495        self.headers.insert(
496            header::CONTENT_TYPE,
497            HeaderValue::from_static("application/json"),
498        );
499        Ok(self)
500    }
501
502    pub fn build(self) -> Result<Request<Body>, Error> {
503        let uri: Uri = self
504            .url
505            .as_str()
506            .parse()
507            .map_err(|error: http::uri::InvalidUri| Error::Header(error.to_string()))?;
508        let mut request = Request::builder()
509            .method(self.method)
510            .uri(uri)
511            .body(self.body)?;
512        *request.headers_mut() = self.headers;
513        Ok(request)
514    }
515
516    pub async fn send(self) -> Result<Response, Error> {
517        let client = self.client.clone();
518        let url = self.url.clone();
519        client.execute_url(url, self.build()?).await
520    }
521}
522
523pub struct Response(HyperResponse<Incoming>);
524
525impl Response {
526    #[must_use]
527    pub fn status(&self) -> StatusCode {
528        self.0.status()
529    }
530
531    #[must_use]
532    pub fn headers(&self) -> &HeaderMap {
533        self.0.headers()
534    }
535
536    pub async fn bytes(self) -> Result<Bytes, Error> {
537        self.0
538            .into_body()
539            .collect()
540            .await
541            .map(|body| body.to_bytes())
542            .map_err(|error| Error::Body(Box::new(error)))
543    }
544
545    pub async fn text(self) -> Result<String, Error> {
546        Ok(String::from_utf8(self.bytes().await?.to_vec())?)
547    }
548
549    pub async fn json<T: DeserializeOwned>(self) -> Result<T, Error> {
550        Ok(serde_json::from_slice(&self.bytes().await?)?)
551    }
552
553    #[must_use]
554    pub fn into_inner(self) -> HyperResponse<Incoming> {
555        self.0
556    }
557}
558
559#[cfg(test)]
560mod tests {
561    use tokio::{
562        io::{AsyncReadExt, AsyncWriteExt},
563        net::TcpListener,
564    };
565
566    use super::*;
567
568    #[test]
569    fn stores_and_scopes_response_cookies() {
570        let url = Url::parse("https://example.com/account/login").unwrap();
571        let mut headers = HeaderMap::new();
572        headers.append(
573            header::SET_COOKIE,
574            HeaderValue::from_static("session=secret; Path=/account; Secure; HttpOnly"),
575        );
576        let mut store = CookieStore::default();
577        store_response_cookies(&mut store, &headers, &url);
578
579        assert_eq!(
580            cookie_header(&store, &url).as_deref(),
581            Some("session=secret")
582        );
583        assert_eq!(
584            cookie_header(&store, &Url::parse("https://example.com/public").unwrap()),
585            None
586        );
587        assert_eq!(
588            cookie_header(&store, &Url::parse("http://example.com/account").unwrap()),
589            None
590        );
591    }
592
593    #[tokio::test]
594    async fn standard_client_uses_tokio_http_connector() {
595        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
596        let address = listener.local_addr().unwrap();
597        let server = tokio::spawn(async move {
598            let (mut stream, _) = listener.accept().await.unwrap();
599            stream
600                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
601                .await
602                .unwrap();
603        });
604
605        let response = standard_client()
606            .get(format!("http://{address}/"))
607            .unwrap()
608            .send()
609            .await
610            .unwrap();
611
612        assert_eq!(response.status(), StatusCode::OK);
613        assert_eq!(response.text().await.unwrap(), "ok");
614        server.await.unwrap();
615    }
616
617    #[tokio::test]
618    async fn standard_client_reuses_pooled_connections() {
619        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
620        let address = listener.local_addr().unwrap();
621        let server = tokio::spawn(async move {
622            let (mut stream, _) = listener.accept().await.unwrap();
623            for _ in 0..2 {
624                let mut request = Vec::new();
625                while !request.ends_with(b"\r\n\r\n") {
626                    let mut byte = [0];
627                    stream.read_exact(&mut byte).await.unwrap();
628                    request.push(byte[0]);
629                }
630                stream
631                    .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
632                    .await
633                    .unwrap();
634            }
635        });
636
637        let client = standard_client_builder()
638            .pool_idle_timeout(Duration::from_secs(30))
639            .pool_max_idle_per_host(1)
640            .build();
641        for _ in 0..2 {
642            let response = client
643                .get(format!("http://{address}/"))
644                .unwrap()
645                .send()
646                .await
647                .unwrap();
648            assert_eq!(response.text().await.unwrap(), "ok");
649        }
650
651        server.await.unwrap();
652    }
653}
rustycrawl_http_client/lib.rs

rustycrawl_http_client/
lib.rs
