rightkit_http/
client.rs

1use std::io::Read;
2use std::sync::Arc;
3use std::time::Duration;
4
5use serde::de::DeserializeOwned;
6use serde::Serialize;
7
8use crate::error::{HttpError, MAX_ERROR_BODY_BYTES};
9use crate::retry::{parse_retry_after, RetryPolicy};
10use crate::sse::SseReader;
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum Method {
14    Get,
15    Post,
16    Put,
17    Patch,
18    Delete,
19    Head,
20}
21
22impl Method {
23    pub fn as_str(self) -> &'static str {
24        match self {
25            Method::Get => "GET",
26            Method::Post => "POST",
27            Method::Put => "PUT",
28            Method::Patch => "PATCH",
29            Method::Delete => "DELETE",
30            Method::Head => "HEAD",
31        }
32    }
33    /// RFC 9110 idempotent methods (safe to retry after an ambiguous failure).
34    pub fn is_idempotent(self) -> bool {
35        !matches!(self, Method::Post | Method::Patch)
36    }
37}
38
39/// One retry decision, reported to [`ClientConfig::on_retry`].
40#[derive(Debug, Clone)]
41pub struct RetryEvent {
42    pub url: String,
43    /// Retry number about to run (1 = first retry).
44    pub retry: u32,
45    pub reason: String,
46    pub delay: Duration,
47}
48
49type RetryObserver = Arc<dyn Fn(&RetryEvent) + Send + Sync>;
50
51#[derive(Clone)]
52pub struct ClientConfig {
53    pub user_agent: String,
54    /// Connection-phase deadline; `None` disables this per-phase deadline.
55    pub connect_timeout: Option<Duration>,
56    /// Whole-request deadline for buffered calls (headers + body).
57    pub request_timeout: Duration,
58    /// Time to receive response headers; for streams this is the first-byte
59    /// (first-token) deadline. `None` disables this per-phase deadline.
60    pub response_timeout: Option<Duration>,
61    /// Total time to receive the body, starting after response headers arrive.
62    /// The budget is not restarted per read. `None` disables this phase deadline.
63    pub recv_body_timeout: Option<Duration>,
64    /// Optional total lifetime cap for a stream body.
65    pub stream_timeout: Option<Duration>,
66    /// Redirects followed (0 = never; credentials are never replayed to a new host).
67    pub max_redirects: u32,
68    /// Error when a nonzero redirect limit is exceeded (default true, like ureq).
69    /// False returns the final 3xx response. With `max_redirects = 0`, redirects
70    /// are never followed and the 3xx response is always returned.
71    pub max_redirects_will_error: bool,
72    pub max_response_bytes: usize,
73    pub retry: RetryPolicy,
74    pub on_retry: Option<RetryObserver>,
75}
76
77impl Default for ClientConfig {
78    fn default() -> Self {
79        Self {
80            user_agent: concat!("rightkit-http/", env!("CARGO_PKG_VERSION")).to_string(),
81            connect_timeout: Some(Duration::from_secs(10)),
82            request_timeout: Duration::from_secs(60),
83            response_timeout: Some(Duration::from_secs(60)),
84            recv_body_timeout: None,
85            stream_timeout: None,
86            max_redirects: 0,
87            max_redirects_will_error: true,
88            max_response_bytes: 32 * 1024 * 1024,
89            retry: RetryPolicy::default(),
90            on_retry: None,
91        }
92    }
93}
94
95#[derive(Debug, Clone)]
96pub struct Request {
97    pub method: Method,
98    pub url: String,
99    pub headers: Vec<(String, String)>,
100    pub body: Vec<u8>,
101    pub timeout: Option<Duration>,
102    pub retry: Option<RetryPolicy>,
103    /// Allow retrying a POST/PATCH after HTTP-status failures (default true for
104    /// status codes; transport errors on non-idempotent methods only retry in
105    /// the connect phase regardless).
106    pub retry_unsafe_statuses: bool,
107}
108
109impl Request {
110    pub fn new(method: Method, url: impl Into<String>) -> Self {
111        Self {
112            method,
113            url: url.into(),
114            headers: Vec::new(),
115            body: Vec::new(),
116            timeout: None,
117            retry: None,
118            retry_unsafe_statuses: true,
119        }
120    }
121    pub fn get(url: impl Into<String>) -> Self {
122        Self::new(Method::Get, url)
123    }
124    pub fn post(url: impl Into<String>) -> Self {
125        Self::new(Method::Post, url)
126    }
127    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
128        self.headers.push((name.into(), value.into()));
129        self
130    }
131    pub fn bearer(self, token: &str) -> Self {
132        self.header("Authorization", format!("Bearer {token}"))
133    }
134    pub fn body(mut self, body: impl Into<Vec<u8>>) -> Self {
135        self.body = body.into();
136        self
137    }
138    pub fn json<T: Serialize>(mut self, value: &T) -> Result<Self, HttpError> {
139        self.body = serde_json::to_vec(value).map_err(|e| HttpError::Json(e.to_string()))?;
140        self.headers
141            .push(("Content-Type".into(), "application/json".into()));
142        Ok(self)
143    }
144    /// `application/x-www-form-urlencoded` body.
145    pub fn form(mut self, pairs: &[(&str, &str)]) -> Self {
146        self.body = form_encode(pairs).into_bytes();
147        self.headers.push((
148            "Content-Type".into(),
149            "application/x-www-form-urlencoded".into(),
150        ));
151        self
152    }
153    pub fn timeout(mut self, d: Duration) -> Self {
154        self.timeout = Some(d);
155        self
156    }
157    pub fn retry(mut self, policy: RetryPolicy) -> Self {
158        self.retry = Some(policy);
159        self
160    }
161}
162
163/// Percent-encode for `application/x-www-form-urlencoded` / URL query values.
164pub fn percent_encode(input: &str) -> String {
165    let mut out = String::with_capacity(input.len());
166    for b in input.bytes() {
167        match b {
168            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
169                out.push(b as char)
170            }
171            b' ' => out.push('+'),
172            _ => out.push_str(&format!("%{b:02X}")),
173        }
174    }
175    out
176}
177
178pub fn percent_decode(input: &str) -> String {
179    let bytes = input.as_bytes();
180    let mut out = Vec::with_capacity(bytes.len());
181    let mut i = 0;
182    while i < bytes.len() {
183        match bytes[i] {
184            b'+' => out.push(b' '),
185            b'%' if i + 2 < bytes.len() => {
186                let hi = (bytes[i + 1] as char).to_digit(16);
187                let lo = (bytes[i + 2] as char).to_digit(16);
188                match (hi, lo) {
189                    (Some(h), Some(l)) => {
190                        out.push((h * 16 + l) as u8);
191                        i += 2;
192                    }
193                    _ => out.push(b'%'),
194                }
195            }
196            b => out.push(b),
197        }
198        i += 1;
199    }
200    String::from_utf8_lossy(&out).into_owned()
201}
202
203pub fn form_encode(pairs: &[(&str, &str)]) -> String {
204    pairs
205        .iter()
206        .map(|(k, v)| format!("{}={}", percent_encode(k), percent_encode(v)))
207        .collect::<Vec<_>>()
208        .join("&")
209}
210
211#[derive(Debug, Clone)]
212pub struct Response {
213    pub status: u16,
214    /// Lower-cased header names.
215    pub headers: Vec<(String, String)>,
216    pub body: Vec<u8>,
217    pub attempts: u32,
218}
219
220impl Response {
221    pub fn header(&self, name: &str) -> Option<&str> {
222        header_of(&self.headers, name)
223    }
224    pub fn text(&self) -> String {
225        String::from_utf8_lossy(&self.body).into_owned()
226    }
227    pub fn json<T: DeserializeOwned>(&self) -> Result<T, HttpError> {
228        serde_json::from_slice(&self.body).map_err(|e| HttpError::Json(e.to_string()))
229    }
230}
231
232pub struct StreamResponse {
233    pub status: u16,
234    pub headers: Vec<(String, String)>,
235    pub attempts: u32,
236    reader: Box<dyn Read + Send>,
237}
238
239impl StreamResponse {
240    pub fn header(&self, name: &str) -> Option<&str> {
241        header_of(&self.headers, name)
242    }
243    pub fn into_reader(self) -> Box<dyn Read + Send> {
244        self.reader
245    }
246    /// Decode the body as Server-Sent Events.
247    pub fn sse(self) -> SseReader<Box<dyn Read + Send>> {
248        SseReader::new(self.reader)
249    }
250}
251
252pub(crate) fn header_of<'a>(headers: &'a [(String, String)], name: &str) -> Option<&'a str> {
253    headers
254        .iter()
255        .find(|(k, _)| k.eq_ignore_ascii_case(name))
256        .map(|(_, v)| v.as_str())
257}
258
259#[derive(Clone)]
260pub struct Client {
261    agent: ureq::Agent,
262    config: ClientConfig,
263}
264
265struct Raw {
266    status: u16,
267    headers: Vec<(String, String)>,
268    body: ureq::Body,
269    attempts: u32,
270}
271
272impl Client {
273    pub fn new(config: ClientConfig) -> Self {
274        let agent: ureq::Agent = ureq::Agent::config_builder()
275            .http_status_as_error(false)
276            .max_redirects(config.max_redirects)
277            .max_redirects_will_error(config.max_redirects_will_error)
278            .user_agent(config.user_agent.clone())
279            .timeout_connect(config.connect_timeout)
280            .build()
281            .into();
282        Self { agent, config }
283    }
284
285    pub fn config(&self) -> &ClientConfig {
286        &self.config
287    }
288
289    /// Buffered request; non-2xx becomes [`HttpError::Status`].
290    pub fn send(&self, req: &Request) -> Result<Response, HttpError> {
291        let resp = self.send_any(req)?;
292        if (200..300).contains(&resp.status) {
293            Ok(resp)
294        } else {
295            Err(status_error(
296                resp.status,
297                &resp.headers,
298                resp.body,
299                resp.attempts,
300            ))
301        }
302    }
303
304    /// Buffered request returning any final status (retryable statuses are still retried).
305    pub fn send_any(&self, req: &Request) -> Result<Response, HttpError> {
306        let timeout = req.timeout.unwrap_or(self.config.request_timeout);
307        let mut raw = self.execute(req, Some(timeout))?;
308        let mut reader = raw.body.as_reader();
309        let body = read_limited(&mut reader, self.config.max_response_bytes)?;
310        Ok(Response {
311            status: raw.status,
312            headers: raw.headers,
313            body,
314            attempts: raw.attempts,
315        })
316    }
317
318    /// Streaming request; non-2xx becomes [`HttpError::Status`]. Retries happen
319    /// only before the response headers arrive (never mid-stream).
320    pub fn stream(&self, req: &Request) -> Result<StreamResponse, HttpError> {
321        let mut raw = self.execute(req, self.config.stream_timeout)?;
322        if !(200..300).contains(&raw.status) {
323            let mut reader = raw.body.as_reader();
324            let body = read_limited(&mut reader, MAX_ERROR_BODY_BYTES).unwrap_or_default();
325            return Err(status_error(raw.status, &raw.headers, body, raw.attempts));
326        }
327        Ok(StreamResponse {
328            status: raw.status,
329            headers: raw.headers,
330            attempts: raw.attempts,
331            reader: Box::new(raw.body.into_reader()),
332        })
333    }
334
335    pub fn get_json<T: DeserializeOwned>(&self, url: &str) -> Result<T, HttpError> {
336        self.send(&Request::get(url).header("Accept", "application/json"))?
337            .json()
338    }
339
340    fn execute(&self, req: &Request, total: Option<Duration>) -> Result<Raw, HttpError> {
341        let policy = req
342            .retry
343            .clone()
344            .unwrap_or_else(|| self.config.retry.clone());
345        let max = policy.max_attempts.max(1);
346        let mut attempt = 0u32;
347        loop {
348            attempt += 1;
349            let outcome = self.once(req, total);
350            let (reason, retry_after) = match outcome {
351                Ok(raw) => {
352                    let retryable = policy.is_retryable_status(raw.status)
353                        && (req.method.is_idempotent() || req.retry_unsafe_statuses);
354                    if !retryable || attempt >= max {
355                        return Ok(Raw {
356                            attempts: attempt,
357                            ..raw
358                        });
359                    }
360                    let ra = header_of(&raw.headers, "retry-after").and_then(parse_retry_after);
361                    // Drain so the connection is released.
362                    let mut body = raw.body;
363                    let mut reader = body.as_reader();
364                    let _ = read_limited(&mut reader, MAX_ERROR_BODY_BYTES);
365                    (format!("HTTP {}", raw.status), ra)
366                }
367                Err(err) => {
368                    let retryable = match &err {
369                        HttpError::Transport { connect_phase, .. } => {
370                            *connect_phase || req.method.is_idempotent()
371                        }
372                        HttpError::Timeout(_) => req.method.is_idempotent(),
373                        _ => false,
374                    };
375                    if !retryable || attempt >= max {
376                        return Err(err);
377                    }
378                    (err.to_string(), None)
379                }
380            };
381            let delay = policy.delay(attempt, retry_after);
382            if let Some(obs) = &self.config.on_retry {
383                obs(&RetryEvent {
384                    url: req.url.clone(),
385                    retry: attempt,
386                    reason,
387                    delay,
388                });
389            }
390            std::thread::sleep(delay);
391        }
392    }
393
394    fn once(&self, req: &Request, total: Option<Duration>) -> Result<Raw, HttpError> {
395        let mut builder = ureq::http::Request::builder()
396            .method(req.method.as_str())
397            .uri(&req.url);
398        for (k, v) in &req.headers {
399            builder = builder.header(k.as_str(), v.as_str());
400        }
401        let request = builder
402            .body(req.body.clone())
403            .map_err(|e| HttpError::InvalidRequest(e.to_string()))?;
404        let configured = self
405            .agent
406            .configure_request(request)
407            .timeout_global(total)
408            .timeout_recv_response(self.config.response_timeout)
409            .timeout_recv_body(self.config.recv_body_timeout)
410            .build();
411        let response = self.agent.run(configured).map_err(map_ureq_error)?;
412        let status = response.status().as_u16();
413        let headers = response
414            .headers()
415            .iter()
416            .map(|(k, v)| {
417                (
418                    k.as_str().to_ascii_lowercase(),
419                    String::from_utf8_lossy(v.as_bytes()).into_owned(),
420                )
421            })
422            .collect();
423        Ok(Raw {
424            status,
425            headers,
426            body: response.into_body(),
427            attempts: 1,
428        })
429    }
430}
431
432pub(crate) fn status_error(
433    status: u16,
434    headers: &[(String, String)],
435    body: Vec<u8>,
436    attempts: u32,
437) -> HttpError {
438    let mut body = body;
439    body.truncate(MAX_ERROR_BODY_BYTES);
440    HttpError::Status {
441        status,
442        body: String::from_utf8_lossy(&body).into_owned(),
443        retry_after: header_of(headers, "retry-after").and_then(parse_retry_after),
444        attempts,
445    }
446}
447
448fn read_limited(reader: &mut dyn Read, limit: usize) -> Result<Vec<u8>, HttpError> {
449    let mut out = Vec::new();
450    reader
451        .take(limit as u64 + 1)
452        .read_to_end(&mut out)
453        .map_err(|e| match e.kind() {
454            std::io::ErrorKind::TimedOut => HttpError::Timeout(e.to_string()),
455            // Body readers wrap ureq errors (including phase timeouts) in
456            // io::ErrorKind::Other. Unwrap them before applying our taxonomy.
457            _ => map_ureq_error(e.into()),
458        })?;
459    if out.len() > limit {
460        return Err(HttpError::BodyTooLarge { limit });
461    }
462    Ok(out)
463}
464
465fn map_ureq_error(err: ureq::Error) -> HttpError {
466    use ureq::Error as E;
467    match err {
468        E::Timeout(t) => HttpError::Timeout(format!("{t:?}")),
469        E::TooManyRedirects => HttpError::Transport {
470            message: "too many redirects".into(),
471            connect_phase: false,
472        },
473        E::HostNotFound => HttpError::Transport {
474            message: "host not found".into(),
475            connect_phase: true,
476        },
477        E::ConnectionFailed => HttpError::Transport {
478            message: "connection failed".into(),
479            connect_phase: true,
480        },
481        E::Io(io)
482            if matches!(
483                io.kind(),
484                std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound
485            ) =>
486        {
487            HttpError::Transport {
488                message: io.to_string(),
489                connect_phase: true,
490            }
491        }
492        E::Io(io) => HttpError::Transport {
493            message: io.to_string(),
494            connect_phase: false,
495        },
496        E::BadUri(m) => HttpError::InvalidRequest(m),
497        E::Http(e) => HttpError::InvalidRequest(e.to_string()),
498        other => HttpError::Transport {
499            message: other.to_string(),
500            connect_phase: false,
501        },
502    }
503}
rightkit_http/client.rs

rightkit_http/
client.rs
