nautilus_network/http/
client.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! HTTP client implementation with rate limiting and timeout support.
17
18use std::{borrow::Cow, collections::HashMap, str::FromStr, sync::Arc, time::Duration};
19
20use nautilus_core::collections::into_ustr_vec;
21use nautilus_cryptography::providers::install_cryptographic_provider;
22use reqwest::{
23    Method, Response, Url,
24    header::{HeaderMap, HeaderName, HeaderValue},
25};
26use ustr::Ustr;
27
28use super::{HttpClientError, HttpResponse, HttpStatus};
29use crate::ratelimiter::{RateLimiter, clock::MonotonicClock, quota::Quota};
30
31/// Default maximum idle connections per host.
32const DEFAULT_POOL_MAX_IDLE_PER_HOST: usize = 32;
33
34/// Default idle connection timeout in seconds.
35const DEFAULT_POOL_IDLE_TIMEOUT_SECS: u64 = 60;
36
37/// Default HTTP/2 keep-alive interval in seconds.
38const DEFAULT_HTTP2_KEEP_ALIVE_SECS: u64 = 30;
39
40/// Default maximum HTTP response body size in bytes (100 MiB).
41///
42/// Bounds peak memory per response so a hostile or malfunctioning endpoint
43/// cannot exhaust memory by streaming an arbitrarily large body. Mirrors the
44/// caps already enforced on the WebSocket and raw-socket paths.
45const DEFAULT_MAX_RESPONSE_BYTES: usize = 100 * 1024 * 1024;
46
47/// An HTTP client that supports rate limiting and timeouts.
48///
49/// Built on `reqwest` for async I/O. Allows per-endpoint and default quotas
50/// through a rate limiter.
51///
52/// This struct is designed to handle HTTP requests efficiently, providing
53/// support for rate limiting, timeouts, and custom headers. The client is
54/// built on top of `reqwest` and can be used for both synchronous and
55/// asynchronous HTTP requests.
56#[derive(Clone, Debug)]
57#[cfg_attr(
58    feature = "python",
59    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.network", from_py_object)
60)]
61#[cfg_attr(
62    feature = "python",
63    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.network")
64)]
65pub struct HttpClient {
66    /// The underlying HTTP client used to make requests.
67    pub(crate) client: InnerHttpClient,
68    /// The rate limiter to control the request rate.
69    pub(crate) rate_limiter: Arc<RateLimiter<Ustr, MonotonicClock>>,
70}
71
72impl HttpClient {
73    /// Creates a new [`HttpClient`] instance.
74    ///
75    /// # Errors
76    ///
77    /// - Returns `InvalidProxy` if the proxy URL is malformed.
78    /// - Returns `ClientBuildError` if building the underlying `reqwest::Client` fails.
79    pub fn new(
80        headers: HashMap<String, String>,
81        header_keys: Vec<String>,
82        keyed_quotas: Vec<(String, Quota)>,
83        default_quota: Option<Quota>,
84        timeout_secs: Option<u64>,
85        proxy_url: Option<String>,
86    ) -> Result<Self, HttpClientError> {
87        let keyed_quotas = keyed_quotas
88            .into_iter()
89            .map(|(key, quota)| (Ustr::from(&key), quota))
90            .collect();
91
92        let rate_limiter = Arc::new(RateLimiter::new_with_quota(default_quota, keyed_quotas));
93
94        Self::new_with_rate_limiter(headers, header_keys, timeout_secs, proxy_url, rate_limiter)
95    }
96
97    /// Creates a new [`HttpClient`] instance sharing an externally-owned rate limiter.
98    ///
99    /// Use this constructor to share a single [`RateLimiter`] across multiple
100    /// [`HttpClient`] instances (for example, the HTTP clients owned by an
101    /// exchange adapter's data and execution clients). All quota state lives
102    /// inside the limiter, so passing the same `Arc` produces a single shared
103    /// bucket.
104    ///
105    /// # Errors
106    ///
107    /// - Returns `InvalidProxy` if the proxy URL is malformed.
108    /// - Returns `ClientBuildError` if building the underlying `reqwest::Client` fails.
109    pub fn new_with_rate_limiter(
110        headers: HashMap<String, String>,
111        header_keys: Vec<String>,
112        timeout_secs: Option<u64>,
113        proxy_url: Option<String>,
114        rate_limiter: Arc<RateLimiter<Ustr, MonotonicClock>>,
115    ) -> Result<Self, HttpClientError> {
116        install_cryptographic_provider();
117
118        // Build default headers
119        let mut header_map = HeaderMap::new();
120
121        for (key, value) in headers {
122            let header_name = HeaderName::from_str(&key)
123                .map_err(|e| HttpClientError::Error(format!("Invalid header name '{key}': {e}")))?;
124            let header_value = HeaderValue::from_str(&value).map_err(|e| {
125                HttpClientError::Error(format!("Invalid header value '{value}': {e}"))
126            })?;
127            header_map.insert(header_name, header_value);
128        }
129
130        let mut client_builder = reqwest::Client::builder()
131            .default_headers(header_map)
132            .tcp_nodelay(true)
133            .pool_max_idle_per_host(DEFAULT_POOL_MAX_IDLE_PER_HOST)
134            .pool_idle_timeout(Duration::from_secs(DEFAULT_POOL_IDLE_TIMEOUT_SECS))
135            .http2_keep_alive_interval(Duration::from_secs(DEFAULT_HTTP2_KEEP_ALIVE_SECS))
136            .http2_keep_alive_while_idle(true)
137            .http2_adaptive_window(true);
138
139        if let Some(timeout_secs) = timeout_secs {
140            client_builder = client_builder.timeout(Duration::from_secs(timeout_secs));
141        }
142
143        // Configure proxy if provided
144        if let Some(proxy_url) = proxy_url {
145            let proxy = reqwest::Proxy::all(&proxy_url)
146                .map_err(|e| HttpClientError::InvalidProxy(format!("{proxy_url}: {e}")))?;
147            client_builder = client_builder.proxy(proxy);
148        }
149
150        let client = client_builder
151            .build()
152            .map_err(|e| HttpClientError::ClientBuildError(e.to_string()))?;
153
154        // Pre-intern header keys as HeaderName, keeping both vectors aligned,
155        // an invalid key is an error: a silent drop would make response extraction read nothing.
156        let (valid_keys, header_names): (Vec<String>, Vec<HeaderName>) = header_keys
157            .into_iter()
158            .map(|k| {
159                HeaderName::from_str(&k)
160                    .map(|name| (k.clone(), name))
161                    .map_err(|e| HttpClientError::Error(format!("Invalid header key '{k}': {e}")))
162            })
163            .collect::<Result<Vec<_>, _>>()?
164            .into_iter()
165            .unzip();
166
167        let client = InnerHttpClient {
168            client,
169            header_keys: Arc::from(valid_keys),
170            header_names: Arc::from(header_names),
171            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
172        };
173
174        Ok(Self {
175            client,
176            rate_limiter,
177        })
178    }
179
180    /// Sends an HTTP request.
181    ///
182    /// # Errors
183    ///
184    /// Returns an error if unable to send request or times out.
185    ///
186    /// # Examples
187    ///
188    /// If requesting `/foo/bar`, pass rate-limit keys `["foo/bar", "foo"]`.
189    #[expect(clippy::too_many_arguments)]
190    pub async fn request(
191        &self,
192        method: Method,
193        url: String,
194        params: Option<&HashMap<String, Vec<String>>>,
195        headers: Option<HashMap<String, String>>,
196        body: Option<Vec<u8>>,
197        timeout_secs: Option<u64>,
198        keys: Option<Vec<String>>,
199    ) -> Result<HttpResponse, HttpClientError> {
200        let keys = keys.map(into_ustr_vec);
201
202        self.request_with_ustr_keys(method, url, params, headers, body, timeout_secs, keys)
203            .await
204    }
205
206    /// Sends an HTTP request with serializable query parameters.
207    ///
208    /// This method accepts any type implementing `Serialize` for query parameters,
209    /// which will be automatically encoded into the URL query string using reqwest's
210    /// `.query()` method, avoiding unnecessary `HashMap` allocations.
211    ///
212    /// # Errors
213    ///
214    /// Returns an error if unable to send request or times out.
215    #[expect(clippy::too_many_arguments)]
216    pub async fn request_with_params<P: serde::Serialize>(
217        &self,
218        method: Method,
219        url: String,
220        params: Option<&P>,
221        headers: Option<HashMap<String, String>>,
222        body: Option<Vec<u8>>,
223        timeout_secs: Option<u64>,
224        keys: Option<Vec<String>>,
225    ) -> Result<HttpResponse, HttpClientError> {
226        let keys = keys.map(into_ustr_vec);
227        let rate_limiter = self.rate_limiter.clone();
228        rate_limiter.await_keys_ready(keys.as_deref()).await;
229
230        self.client
231            .send_request_with_query(method, url, params, headers, body, timeout_secs)
232            .await
233    }
234
235    /// Sends an HTTP request using pre-interned rate limiter keys.
236    ///
237    /// # Errors
238    ///
239    /// Returns an error if unable to send the request or the request times out.
240    #[expect(clippy::too_many_arguments)]
241    pub async fn request_with_ustr_keys(
242        &self,
243        method: Method,
244        url: String,
245        params: Option<&HashMap<String, Vec<String>>>,
246        headers: Option<HashMap<String, String>>,
247        body: Option<Vec<u8>>,
248        timeout_secs: Option<u64>,
249        keys: Option<Vec<Ustr>>,
250    ) -> Result<HttpResponse, HttpClientError> {
251        let rate_limiter = self.rate_limiter.clone();
252        rate_limiter.await_keys_ready(keys.as_deref()).await;
253
254        self.client
255            .send_request(method, url, params, headers, body, timeout_secs)
256            .await
257    }
258
259    /// Sends an HTTP GET request.
260    ///
261    /// # Errors
262    ///
263    /// Returns an error if unable to send request or times out.
264    pub async fn get(
265        &self,
266        url: String,
267        params: Option<&HashMap<String, Vec<String>>>,
268        headers: Option<HashMap<String, String>>,
269        timeout_secs: Option<u64>,
270        keys: Option<Vec<String>>,
271    ) -> Result<HttpResponse, HttpClientError> {
272        self.request(Method::GET, url, params, headers, None, timeout_secs, keys)
273            .await
274    }
275
276    /// Sends an HTTP POST request.
277    ///
278    /// # Errors
279    ///
280    /// Returns an error if unable to send request or times out.
281    pub async fn post(
282        &self,
283        url: String,
284        params: Option<&HashMap<String, Vec<String>>>,
285        headers: Option<HashMap<String, String>>,
286        body: Option<Vec<u8>>,
287        timeout_secs: Option<u64>,
288        keys: Option<Vec<String>>,
289    ) -> Result<HttpResponse, HttpClientError> {
290        self.request(Method::POST, url, params, headers, body, timeout_secs, keys)
291            .await
292    }
293
294    /// Sends an HTTP PATCH request.
295    ///
296    /// # Errors
297    ///
298    /// Returns an error if unable to send request or times out.
299    pub async fn patch(
300        &self,
301        url: String,
302        params: Option<&HashMap<String, Vec<String>>>,
303        headers: Option<HashMap<String, String>>,
304        body: Option<Vec<u8>>,
305        timeout_secs: Option<u64>,
306        keys: Option<Vec<String>>,
307    ) -> Result<HttpResponse, HttpClientError> {
308        self.request(
309            Method::PATCH,
310            url,
311            params,
312            headers,
313            body,
314            timeout_secs,
315            keys,
316        )
317        .await
318    }
319
320    /// Sends an HTTP DELETE request.
321    ///
322    /// # Errors
323    ///
324    /// Returns an error if unable to send request or times out.
325    pub async fn delete(
326        &self,
327        url: String,
328        params: Option<&HashMap<String, Vec<String>>>,
329        headers: Option<HashMap<String, String>>,
330        timeout_secs: Option<u64>,
331        keys: Option<Vec<String>>,
332    ) -> Result<HttpResponse, HttpClientError> {
333        self.request(
334            Method::DELETE,
335            url,
336            params,
337            headers,
338            None,
339            timeout_secs,
340            keys,
341        )
342        .await
343    }
344}
345
346/// Internal implementation backing [`HttpClient`].
347///
348/// The client is backed by a [`reqwest::Client`] which keeps connections alive and
349/// can be cloned cheaply. The client also has a list of header fields to
350/// extract from the response.
351///
352/// The client returns an [`HttpResponse`]. The client filters only the key value
353/// for the give `header_keys`.
354#[derive(Clone, Debug)]
355pub struct InnerHttpClient {
356    pub(crate) client: reqwest::Client,
357    pub(crate) header_keys: Arc<[String]>,
358    pub(crate) header_names: Arc<[HeaderName]>,
359    /// Maximum response body size in bytes; bodies exceeding this are rejected.
360    pub(crate) max_response_bytes: usize,
361}
362
363impl InnerHttpClient {
364    /// Sends an HTTP request and returns an [`HttpResponse`].
365    ///
366    /// # Errors
367    ///
368    /// Returns an error if unable to send request or times out.
369    pub async fn send_request(
370        &self,
371        method: Method,
372        url: String,
373        params: Option<&HashMap<String, Vec<String>>>,
374        headers: Option<HashMap<String, String>>,
375        body: Option<Vec<u8>>,
376        timeout_secs: Option<u64>,
377    ) -> Result<HttpResponse, HttpClientError> {
378        let full_url = encode_url_params(&url, params)?;
379        self.send_request_internal(
380            method,
381            full_url.as_ref(),
382            None::<&()>,
383            headers,
384            body,
385            timeout_secs,
386        )
387        .await
388    }
389
390    /// Sends an HTTP request with query parameters using reqwest's `.query()` method.
391    ///
392    /// This method accepts any type implementing `Serialize` for query parameters,
393    /// avoiding `HashMap` conversion overhead.
394    ///
395    /// # Errors
396    ///
397    /// Returns an error if unable to send request or times out.
398    pub async fn send_request_with_query<Q: serde::Serialize>(
399        &self,
400        method: Method,
401        url: String,
402        query: Option<&Q>,
403        headers: Option<HashMap<String, String>>,
404        body: Option<Vec<u8>>,
405        timeout_secs: Option<u64>,
406    ) -> Result<HttpResponse, HttpClientError> {
407        self.send_request_internal(method, &url, query, headers, body, timeout_secs)
408            .await
409    }
410
411    /// Internal implementation for sending HTTP requests.
412    ///
413    /// # Errors
414    ///
415    /// Returns an error if unable to send request or times out.
416    async fn send_request_internal<Q: serde::Serialize>(
417        &self,
418        method: Method,
419        url: &str,
420        query: Option<&Q>,
421        headers: Option<HashMap<String, String>>,
422        body: Option<Vec<u8>>,
423        timeout_secs: Option<u64>,
424    ) -> Result<HttpResponse, HttpClientError> {
425        let reqwest_url =
426            Url::parse(url).map_err(|e| HttpClientError::from(format!("URL parse error: {e}")))?;
427
428        let mut request_builder = self.client.request(method, reqwest_url);
429
430        if let Some(headers) = headers {
431            let mut header_map = HeaderMap::with_capacity(headers.len());
432            for (header_key, header_value) in &headers {
433                let key = HeaderName::from_bytes(header_key.as_bytes())
434                    .map_err(|e| HttpClientError::from(format!("Invalid header name: {e}")))?;
435
436                if let Some(old_value) = header_map.insert(
437                    key.clone(),
438                    header_value
439                        .parse()
440                        .map_err(|e| HttpClientError::from(format!("Invalid header value: {e}")))?,
441                ) {
442                    log::trace!("Replaced header '{key}': old={old_value:?}, new={header_value}");
443                }
444            }
445            request_builder = request_builder.headers(header_map);
446        }
447
448        if let Some(q) = query {
449            request_builder = request_builder.query(q);
450        }
451
452        if let Some(timeout_secs) = timeout_secs {
453            request_builder = request_builder.timeout(Duration::new(timeout_secs, 0));
454        }
455
456        let request = match body {
457            Some(b) => request_builder
458                .body(b)
459                .build()
460                .map_err(HttpClientError::from)?,
461            None => request_builder.build().map_err(HttpClientError::from)?,
462        };
463
464        log::trace!("{} {}", request.method(), request.url());
465
466        let response = self
467            .client
468            .execute(request)
469            .await
470            .map_err(HttpClientError::from)?;
471
472        self.to_response(response).await
473    }
474
475    /// Converts a `reqwest::Response` into an `HttpResponse`.
476    ///
477    /// Uses pre-interned `HeaderName` values to avoid string-to-header parsing per response.
478    ///
479    /// # Errors
480    ///
481    /// Returns an error if unable to send request or times out.
482    pub async fn to_response(&self, response: Response) -> Result<HttpResponse, HttpClientError> {
483        log::trace!("{response:?}");
484
485        let resp_headers = response.headers();
486        let mut headers =
487            HashMap::with_capacity(std::cmp::min(self.header_names.len(), resp_headers.len()));
488
489        for (name, key_str) in self.header_names.iter().zip(self.header_keys.iter()) {
490            if let Some(val) = resp_headers.get(name)
491                && let Ok(v) = val.to_str()
492            {
493                headers.insert(key_str.clone(), v.to_owned());
494            }
495        }
496
497        let status = HttpStatus::new(response.status());
498        let body = self.read_body_capped(response).await?;
499
500        Ok(HttpResponse {
501            status,
502            headers,
503            body,
504        })
505    }
506
507    /// Reads the response body, rejecting any body that exceeds `max_response_bytes`.
508    ///
509    /// A `Content-Length` larger than the cap is rejected up front; otherwise the
510    /// body is streamed chunk-by-chunk and aborted as soon as the accumulated size
511    /// would exceed the cap, so an oversized or unbounded (chunked) body is never
512    /// fully buffered into memory.
513    ///
514    /// # Errors
515    ///
516    /// Returns an error if the body exceeds the configured maximum size, or if
517    /// reading a chunk fails.
518    async fn read_body_capped(
519        &self,
520        mut response: Response,
521    ) -> Result<bytes::Bytes, HttpClientError> {
522        let max = self.max_response_bytes;
523
524        // Fast path: reject up front when the advertised length already exceeds the cap.
525        if let Some(len) = response.content_length()
526            && len > max as u64
527        {
528            return Err(HttpClientError::Error(format!(
529                "HTTP response body of {len} bytes exceeds maximum of {max} bytes",
530            )));
531        }
532
533        let mut buf = bytes::BytesMut::new();
534        while let Some(chunk) = response.chunk().await.map_err(HttpClientError::from)? {
535            if buf.len() + chunk.len() > max {
536                return Err(HttpClientError::Error(format!(
537                    "HTTP response body exceeds maximum of {max} bytes",
538                )));
539            }
540            buf.extend_from_slice(&chunk);
541        }
542
543        Ok(buf.freeze())
544    }
545}
546
547impl Default for InnerHttpClient {
548    /// Creates a new default [`InnerHttpClient`] instance.
549    ///
550    /// The default client is initialized with an empty list of header keys and a new `reqwest::Client`.
551    fn default() -> Self {
552        install_cryptographic_provider();
553        let client = reqwest::Client::new();
554        Self {
555            client,
556            header_keys: Arc::default(),
557            header_names: Arc::default(),
558            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
559        }
560    }
561}
562
563/// Encodes URL parameters into the query string.
564///
565/// Returns `Cow::Borrowed` when no parameters need appending (zero-alloc fast path).
566/// Parameters can have multiple values per key (for doseq=True behavior).
567/// Preserves existing query strings in the URL by appending with '&' instead of '?'.
568fn encode_url_params<'a>(
569    url: &'a str,
570    params: Option<&HashMap<String, Vec<String>>>,
571) -> Result<Cow<'a, str>, HttpClientError> {
572    let Some(params) = params else {
573        return Ok(Cow::Borrowed(url));
574    };
575
576    let pairs: Vec<(&str, &str)> = params
577        .iter()
578        .flat_map(|(key, values)| {
579            values
580                .iter()
581                .map(move |value| (key.as_str(), value.as_str()))
582        })
583        .collect();
584
585    if pairs.is_empty() {
586        return Ok(Cow::Borrowed(url));
587    }
588
589    let query_string = serde_urlencoded::to_string(pairs)
590        .map_err(|e| HttpClientError::Error(format!("Failed to encode params: {e}")))?;
591
592    let separator = if url.contains('?') { '&' } else { '?' };
593    Ok(Cow::Owned(format!("{url}{separator}{query_string}")))
594}
595
596#[cfg(test)]
597#[cfg(target_os = "linux")] // Only run network tests on Linux (CI stability)
598mod tests {
599    use std::net::SocketAddr;
600
601    use axum::{
602        Router,
603        routing::{delete, get, patch, post},
604        serve,
605    };
606    use http::status::StatusCode;
607    use rstest::rstest;
608
609    use super::*;
610
611    fn create_router() -> Router {
612        Router::new()
613            .route("/get", get(|| async { "hello-world!" }))
614            .route("/post", post(|| async { StatusCode::OK }))
615            .route("/patch", patch(|| async { StatusCode::OK }))
616            .route("/delete", delete(|| async { StatusCode::OK }))
617            .route("/notfound", get(|| async { StatusCode::NOT_FOUND }))
618            .route(
619                "/slow",
620                get(|| async {
621                    tokio::time::sleep(Duration::from_secs(2)).await;
622                    "Eventually responded"
623                }),
624            )
625            .route(
626                "/large",
627                // Returns a 1 MiB body to exercise the response size cap.
628                get(|| async { "x".repeat(1024 * 1024) }),
629            )
630    }
631
632    async fn start_test_server() -> Result<SocketAddr, Box<dyn std::error::Error + Send + Sync>> {
633        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
634        let addr = listener.local_addr().unwrap();
635
636        tokio::spawn(async move {
637            serve(listener, create_router()).await.unwrap();
638        });
639
640        Ok(addr)
641    }
642
643    #[tokio::test]
644    async fn test_get() {
645        let addr = start_test_server().await.unwrap();
646        let url = format!("http://{addr}");
647
648        let client = InnerHttpClient::default();
649        let response = client
650            .send_request(
651                reqwest::Method::GET,
652                format!("{url}/get"),
653                None,
654                None,
655                None,
656                None,
657            )
658            .await
659            .unwrap();
660
661        assert!(response.status.is_success());
662        assert_eq!(String::from_utf8_lossy(&response.body), "hello-world!");
663    }
664
665    #[tokio::test]
666    async fn test_response_body_within_cap_is_returned() {
667        let addr = start_test_server().await.unwrap();
668        let url = format!("http://{addr}");
669
670        // Cap above the 1 MiB payload: body should be returned intact.
671        let client = InnerHttpClient {
672            max_response_bytes: 4 * 1024 * 1024,
673            ..Default::default()
674        };
675
676        let response = client
677            .send_request(
678                reqwest::Method::GET,
679                format!("{url}/large"),
680                None,
681                None,
682                None,
683                None,
684            )
685            .await
686            .unwrap();
687
688        assert!(response.status.is_success());
689        assert_eq!(response.body.len(), 1024 * 1024);
690    }
691
692    #[tokio::test]
693    async fn test_response_body_exceeding_cap_is_rejected() {
694        let addr = start_test_server().await.unwrap();
695        let url = format!("http://{addr}");
696
697        // Cap below the 1 MiB payload: the request must fail rather than buffer it.
698        let client = InnerHttpClient {
699            max_response_bytes: 16 * 1024,
700            ..Default::default()
701        };
702
703        let result = client
704            .send_request(
705                reqwest::Method::GET,
706                format!("{url}/large"),
707                None,
708                None,
709                None,
710                None,
711            )
712            .await;
713
714        let err = result.expect_err("oversized response body should be rejected");
715        assert!(
716            err.to_string().contains("exceeds maximum"),
717            "unexpected error: {err}",
718        );
719    }
720
721    #[tokio::test]
722    async fn test_post() {
723        let addr = start_test_server().await.unwrap();
724        let url = format!("http://{addr}");
725
726        let client = InnerHttpClient::default();
727        let response = client
728            .send_request(
729                reqwest::Method::POST,
730                format!("{url}/post"),
731                None,
732                None,
733                None,
734                None,
735            )
736            .await
737            .unwrap();
738
739        assert!(response.status.is_success());
740    }
741
742    #[tokio::test]
743    async fn test_post_with_body() {
744        let addr = start_test_server().await.unwrap();
745        let url = format!("http://{addr}");
746
747        let client = InnerHttpClient::default();
748
749        let mut body = HashMap::new();
750        body.insert(
751            "key1".to_string(),
752            serde_json::Value::String("value1".to_string()),
753        );
754        body.insert(
755            "key2".to_string(),
756            serde_json::Value::String("value2".to_string()),
757        );
758
759        let body_string = serde_json::to_string(&body).unwrap();
760        let body_bytes = body_string.into_bytes();
761
762        let response = client
763            .send_request(
764                reqwest::Method::POST,
765                format!("{url}/post"),
766                None,
767                None,
768                Some(body_bytes),
769                None,
770            )
771            .await
772            .unwrap();
773
774        assert!(response.status.is_success());
775    }
776
777    #[tokio::test]
778    async fn test_patch() {
779        let addr = start_test_server().await.unwrap();
780        let url = format!("http://{addr}");
781
782        let client = InnerHttpClient::default();
783        let response = client
784            .send_request(
785                reqwest::Method::PATCH,
786                format!("{url}/patch"),
787                None,
788                None,
789                None,
790                None,
791            )
792            .await
793            .unwrap();
794
795        assert!(response.status.is_success());
796    }
797
798    #[tokio::test]
799    async fn test_delete() {
800        let addr = start_test_server().await.unwrap();
801        let url = format!("http://{addr}");
802
803        let client = InnerHttpClient::default();
804        let response = client
805            .send_request(
806                reqwest::Method::DELETE,
807                format!("{url}/delete"),
808                None,
809                None,
810                None,
811                None,
812            )
813            .await
814            .unwrap();
815
816        assert!(response.status.is_success());
817    }
818
819    #[tokio::test]
820    async fn test_not_found() {
821        let addr = start_test_server().await.unwrap();
822        let url = format!("http://{addr}/notfound");
823        let client = InnerHttpClient::default();
824
825        let response = client
826            .send_request(reqwest::Method::GET, url, None, None, None, None)
827            .await
828            .unwrap();
829
830        assert!(response.status.is_client_error());
831        assert_eq!(response.status.as_u16(), 404);
832    }
833
834    #[tokio::test]
835    async fn test_timeout() {
836        let addr = start_test_server().await.unwrap();
837        let url = format!("http://{addr}/slow");
838        let client = InnerHttpClient::default();
839
840        // We'll set a 1-second timeout for a route that sleeps 2 seconds
841        let result = client
842            .send_request(reqwest::Method::GET, url, None, None, None, Some(1))
843            .await;
844
845        match result {
846            Err(HttpClientError::TimeoutError(msg)) => {
847                println!("Got expected timeout error: {msg}");
848            }
849            Err(e) => panic!("Expected a timeout error, was: {e:?}"),
850            Ok(resp) => panic!("Expected a timeout error, but was a successful response: {resp:?}"),
851        }
852    }
853
854    #[rstest]
855    fn test_http_client_without_proxy() {
856        // Create client with no proxy
857        let result = HttpClient::new(
858            HashMap::new(),
859            vec![],
860            vec![],
861            None,
862            None,
863            None, // No proxy
864        );
865
866        assert!(result.is_ok());
867    }
868
869    #[rstest]
870    fn test_http_client_with_valid_proxy() {
871        // Create client with a valid proxy URL
872        let result = HttpClient::new(
873            HashMap::new(),
874            vec![],
875            vec![],
876            None,
877            None,
878            Some("http://proxy.example.com:8080".to_string()),
879        );
880
881        assert!(result.is_ok());
882    }
883
884    #[rstest]
885    fn test_http_client_with_socks5_proxy() {
886        // Create client with a SOCKS5 proxy URL
887        let result = HttpClient::new(
888            HashMap::new(),
889            vec![],
890            vec![],
891            None,
892            None,
893            Some("socks5://127.0.0.1:1080".to_string()),
894        );
895
896        assert!(result.is_ok());
897    }
898
899    #[rstest]
900    fn test_http_client_with_malformed_proxy() {
901        // Note: reqwest::Proxy::all() is lenient and accepts most strings.
902        // It only fails on obviously malformed URLs like "://invalid" or "http://".
903        // More subtle issues (like "not-a-valid-url") are caught when connecting.
904        let result = HttpClient::new(
905            HashMap::new(),
906            vec![],
907            vec![],
908            None,
909            None,
910            Some("://invalid".to_string()),
911        );
912
913        assert!(result.is_err());
914        assert!(matches!(result, Err(HttpClientError::InvalidProxy(_))));
915    }
916
917    #[rstest]
918    fn test_http_client_with_empty_proxy_string() {
919        // Create client with an empty proxy URL string
920        let result = HttpClient::new(
921            HashMap::new(),
922            vec![],
923            vec![],
924            None,
925            None,
926            Some(String::new()),
927        );
928
929        assert!(result.is_err());
930        assert!(matches!(result, Err(HttpClientError::InvalidProxy(_))));
931    }
932
933    #[tokio::test]
934    async fn test_http_client_get() {
935        let addr = start_test_server().await.unwrap();
936        let url = format!("http://{addr}/get");
937
938        let client = HttpClient::new(HashMap::new(), vec![], vec![], None, None, None).unwrap();
939        let response = client.get(url, None, None, None, None).await.unwrap();
940
941        assert!(response.status.is_success());
942        assert_eq!(String::from_utf8_lossy(&response.body), "hello-world!");
943    }
944
945    #[tokio::test]
946    async fn test_http_client_post() {
947        let addr = start_test_server().await.unwrap();
948        let url = format!("http://{addr}/post");
949
950        let client = HttpClient::new(HashMap::new(), vec![], vec![], None, None, None).unwrap();
951        let response = client
952            .post(url, None, None, None, None, None)
953            .await
954            .unwrap();
955
956        assert!(response.status.is_success());
957    }
958
959    #[tokio::test]
960    async fn test_http_client_patch() {
961        let addr = start_test_server().await.unwrap();
962        let url = format!("http://{addr}/patch");
963
964        let client = HttpClient::new(HashMap::new(), vec![], vec![], None, None, None).unwrap();
965        let response = client
966            .patch(url, None, None, None, None, None)
967            .await
968            .unwrap();
969
970        assert!(response.status.is_success());
971    }
972
973    #[tokio::test]
974    async fn test_http_client_delete() {
975        let addr = start_test_server().await.unwrap();
976        let url = format!("http://{addr}/delete");
977
978        let client = HttpClient::new(HashMap::new(), vec![], vec![], None, None, None).unwrap();
979        let response = client.delete(url, None, None, None, None).await.unwrap();
980
981        assert!(response.status.is_success());
982    }
983}
nautilus_network/http/client.rs

nautilus_network/http/
client.rs
