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

rustycrawl_http_client/
lib.rs
