jules_api/http/
reqwest_transport.rs

1//! A [`Transport`] implementation backed by [`reqwest`], for non-wasm targets.
2
3use super::{HttpRequest, HttpResponse, Method, Transport};
4use jules_core::errors::{NetworkError, SDKError};
5use std::future::Future;
6use std::time::Duration;
7
8/// A [`Transport`] implementation that performs real network requests using
9/// [`reqwest`].
10///
11/// This is the only place in the SDK where a request timeout is actually
12/// enforced.
13#[derive(Debug, Clone)]
14pub struct ReqwestTransport {
15    client: reqwest::Client,
16}
17
18impl ReqwestTransport {
19    /// Creates a new `ReqwestTransport` whose underlying HTTP client enforces
20    /// the given request `timeout`.
21    ///
22    /// # Panics
23    ///
24    /// Panics if the underlying `reqwest::Client` fails to build (e.g. if the
25    /// TLS backend cannot be initialized).
26    #[must_use]
27    pub fn new(timeout: Duration) -> Self {
28        let client = reqwest::Client::builder()
29            .timeout(timeout)
30            .build()
31            .expect("failed to build reqwest client");
32        Self { client }
33    }
34}
35
36fn method_to_reqwest(method: Method) -> reqwest::Method {
37    match method {
38        Method::Get => reqwest::Method::GET,
39        Method::Post => reqwest::Method::POST,
40        Method::Put => reqwest::Method::PUT,
41        Method::Delete => reqwest::Method::DELETE,
42        Method::Patch => reqwest::Method::PATCH,
43    }
44}
45
46fn map_reqwest_error(err: &reqwest::Error) -> SDKError {
47    let message = if err.is_timeout() {
48        format!("request timed out: {err}")
49    } else {
50        format!("request failed: {err}")
51    };
52    SDKError::Network(NetworkError::new(message))
53}
54
55impl Transport for ReqwestTransport {
56    fn send(
57        &self,
58        request: HttpRequest,
59    ) -> impl Future<Output = Result<HttpResponse, SDKError>> + Send {
60        let client = self.client.clone();
61        async move {
62            let mut builder = client.request(method_to_reqwest(request.method), &request.url);
63            for (key, value) in &request.headers {
64                builder = builder.header(key, value);
65            }
66            if let Some(body) = request.body {
67                builder = builder.body(body);
68            }
69
70            let response = builder.send().await.map_err(|e| map_reqwest_error(&e))?;
71
72            let status = response.status().as_u16();
73            let headers = response
74                .headers()
75                .iter()
76                .map(|(name, value)| {
77                    (
78                        name.to_string(),
79                        value.to_str().unwrap_or_default().to_string(),
80                    )
81                })
82                .collect();
83            let body = response
84                .bytes()
85                .await
86                .map_err(|e| map_reqwest_error(&e))?
87                .to_vec();
88
89            Ok(HttpResponse::new(status, headers, body))
90        }
91    }
92}
93
94#[cfg(test)]
95mod tests {
96    use super::*;
97    use tokio::io::{AsyncReadExt, AsyncWriteExt};
98    use tokio::net::TcpListener;
99
100    #[tokio::test]
101    async fn test_send_real_http_response() {
102        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
103        let addr = listener.local_addr().unwrap();
104
105        tokio::spawn(async move {
106            let (mut socket, _) = listener.accept().await.unwrap();
107            // Drain the request so the client isn't blocked writing it.
108            let mut buf = [0u8; 1024];
109            let _ = socket.read(&mut buf).await;
110
111            let body = b"hello world";
112            let response = format!(
113                "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
114                body.len()
115            );
116            socket.write_all(response.as_bytes()).await.unwrap();
117            socket.write_all(body).await.unwrap();
118            socket.shutdown().await.unwrap();
119        });
120
121        let transport = ReqwestTransport::new(Duration::from_secs(5));
122        let request = HttpRequest::new(Method::Get, format!("http://{addr}/test"));
123        let response = transport.send(request).await.unwrap();
124
125        assert_eq!(response.status, 200);
126        assert_eq!(response.body, b"hello world");
127        assert!(response
128            .headers
129            .iter()
130            .any(|(k, v)| k.eq_ignore_ascii_case("content-type") && v == "text/plain"));
131    }
132
133    #[tokio::test]
134    async fn test_send_enforces_timeout() {
135        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
136        let addr = listener.local_addr().unwrap();
137
138        tokio::spawn(async move {
139            // Accept the connection but never respond, holding it open.
140            let (_socket, _) = listener.accept().await.unwrap();
141            tokio::time::sleep(Duration::from_secs(10)).await;
142        });
143
144        let transport = ReqwestTransport::new(Duration::from_millis(100));
145        let request = HttpRequest::new(Method::Get, format!("http://{addr}/slow"));
146
147        let result = tokio::time::timeout(Duration::from_secs(2), transport.send(request))
148            .await
149            .expect("send() did not return within the bounded time");
150
151        match result {
152            Err(SDKError::Network(err)) => {
153                assert!(
154                    err.message.to_lowercase().contains("timed out")
155                        || err.message.to_lowercase().contains("timeout"),
156                    "expected timeout message, got: {}",
157                    err.message
158                );
159            }
160            other => panic!("expected SDKError::Network due to timeout, got: {other:?}"),
161        }
162    }
163}
jules_api/http/reqwest_transport.rs

jules_api/http/
reqwest_transport.rs
