acorn/io/http/
mod.rs

1//! HTTP client utilities for ACORN IO operations
2use super::ApiResult;
3use crate::util::Label;
4use async_trait::async_trait;
5use axum::http::HeaderMap;
6use chrono::Utc;
7use color_eyre::eyre::eyre;
8use core::fmt;
9use reqwest::header::USER_AGENT;
10use serde::{Deserialize, Serialize};
11use tower::{service_fn, ServiceExt};
12use tracing::{debug, warn};
13
14pub mod policy;
15
16/// See recommendation at <https://support.datacite.org/docs/api#api-versions>
17#[cfg(feature = "std")]
18const ACORN_USER_AGENT: &str = concat!("ACORN/", env!("CARGO_PKG_VERSION"), " (https://acorn.ornl.gov; mailto:research@ornl.gov)");
19#[cfg(not(feature = "std"))]
20const ACORN_USER_AGENT: &str = "ACORN (https://acorn.ornl.gov; mailto:research@ornl.gov)";
21/// Async HTTP service abstraction for ACORN I/O.
22#[async_trait]
23pub trait HttpService {
24    /// Execute an HTTP request.
25    async fn execute(&self, request: HttpRequest) -> ApiResult<HttpResponse>;
26}
27/// Supported HTTP methods for ACORN I/O requests.
28#[derive(Clone, Debug, Default, Deserialize, Serialize)]
29#[serde(rename_all = "lowercase")]
30pub enum HttpMethod {
31    /// HTTP GET
32    #[default]
33    Get,
34    /// HTTP DELETE
35    Delete,
36    /// HTTP PATCH
37    Patch,
38    /// HTTP POST
39    Post,
40    /// HTTP PUT
41    Put,
42}
43/// Internal HTTP request representation.
44#[derive(Clone, Debug)]
45pub struct HttpRequest {
46    /// Request headers
47    pub headers: HeaderMap,
48    /// Optional JSON body
49    pub json_body: Option<serde_json::Value>,
50    /// HTTP method
51    pub method: HttpMethod,
52    /// URL string
53    pub url: String,
54}
55/// Fluent async request builder backed by [`HttpService`].
56#[derive(Clone, Debug)]
57pub struct HttpRequestBuilder {
58    request: HttpRequest,
59    service: ReqwestHttpService,
60}
61/// Internal HTTP response representation.
62#[derive(Clone, Debug)]
63pub struct HttpResponse {
64    /// Response body bytes
65    pub body: Vec<u8>,
66    /// Response headers
67    pub headers: HeaderMap,
68    /// HTTP status code
69    pub status_code: u16,
70}
71/// Reqwest-backed async HTTP service adapter.
72#[derive(Clone, Debug)]
73pub struct ReqwestHttpService {
74    client: reqwest::Client,
75}
76impl Default for ReqwestHttpService {
77    fn default() -> Self {
78        let policy = policy::shared_http_policy();
79        let client = reqwest::Client::builder()
80            .timeout(policy.timeout)
81            .connect_timeout(policy.connect_timeout)
82            .build()
83            .unwrap_or_else(|_| reqwest::Client::new());
84        Self { client }
85    }
86}
87impl From<&str> for HttpMethod {
88    fn from(value: &str) -> Self {
89        match value.to_uppercase().as_str() {
90            | "DELETE" => HttpMethod::Delete,
91            | "GET" => HttpMethod::Get,
92            | "PATCH" => HttpMethod::Patch,
93            | "POST" => HttpMethod::Post,
94            | "PUT" => HttpMethod::Put,
95            | _ => HttpMethod::Get,
96        }
97    }
98}
99impl From<HttpMethod> for reqwest::Method {
100    fn from(value: HttpMethod) -> Self {
101        match value {
102            | HttpMethod::Delete => reqwest::Method::DELETE,
103            | HttpMethod::Get => reqwest::Method::GET,
104            | HttpMethod::Patch => reqwest::Method::PATCH,
105            | HttpMethod::Post => reqwest::Method::POST,
106            | HttpMethod::Put => reqwest::Method::PUT,
107        }
108    }
109}
110impl fmt::Display for HttpRequestBuilder {
111    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
112        write!(f, "{} {}", self.request.method, self.request.url)
113    }
114}
115impl HttpRequestBuilder {
116    /// Add headers to the request.
117    pub fn headers(mut self, headers: HeaderMap) -> Self {
118        self.request.headers.extend(headers);
119        self
120    }
121    /// Add a JSON body to the request.
122    pub fn json(mut self, value: &serde_json::Value) -> Self {
123        self.request.json_body = Some(value.clone());
124        self
125    }
126    fn new(method: HttpMethod, url: impl Into<String>) -> Self {
127        Self {
128            request: HttpRequest {
129                headers: HeaderMap::new(),
130                json_body: None,
131                method,
132                url: url.into(),
133            },
134            service: ReqwestHttpService::default(),
135        }
136    }
137    /// Send the request.
138    pub async fn send(self) -> ApiResult<HttpResponse> {
139        self.service.execute(self.request).await
140    }
141}
142impl HttpResponse {
143    /// Read response body as bytes.
144    pub async fn bytes(self) -> ApiResult<Vec<u8>> {
145        Ok(self.body)
146    }
147    /// Read response body as text.
148    pub async fn text(self) -> ApiResult<String> {
149        match String::from_utf8(self.body) {
150            | Ok(value) => Ok(value),
151            | Err(why) => Err(eyre!("HTTP response body is not valid UTF-8 — {why}")),
152        }
153    }
154}
155#[async_trait]
156impl HttpService for ReqwestHttpService {
157    async fn execute(&self, request: HttpRequest) -> ApiResult<HttpResponse> {
158        execute_with_policy(self.client.clone(), request).await
159    }
160}
161async fn execute_with_policy(client: reqwest::Client, request: HttpRequest) -> ApiResult<HttpResponse> {
162    let policy = policy::shared_http_policy();
163    let method = request.method.clone();
164    let url = request.url.clone();
165    let max_attempts = policy.max_attempts();
166    for attempt in 1..=max_attempts {
167        let started = Utc::now();
168        let result = execute_with_timeout(client.clone(), request.clone()).await;
169        #[allow(clippy::arithmetic_side_effects)]
170        let elapsed_ms = (Utc::now() - started).num_milliseconds();
171        match result {
172            | Ok(response) => {
173                let retry = should_retry(&method, Some(response.status_code));
174                if retry && attempt < max_attempts {
175                    warn!(
176                        attempt,
177                        status_code = response.status_code,
178                        elapsed_ms,
179                        url,
180                        "=> {} Retrying HTTP request",
181                        Label::using()
182                    );
183                } else {
184                    debug!(
185                        attempt,
186                        status_code = response.status_code,
187                        elapsed_ms,
188                        url,
189                        "=> {} HTTP request",
190                        Label::using()
191                    );
192                    return Ok(response);
193                }
194            }
195            | Err(why) => {
196                let retry = should_retry(&method, None);
197                if retry && attempt < max_attempts {
198                    warn!(attempt, elapsed_ms, url, "=> {} Retrying HTTP request — {why}", Label::using());
199                } else {
200                    warn!(attempt, elapsed_ms, url, "=> {} HTTP request failed — {why}", Label::fail());
201                    return Err(why);
202                }
203            }
204        }
205    }
206    Err(eyre!("HTTP request failed after retry attempts"))
207}
208async fn execute_with_timeout(client: reqwest::Client, request: HttpRequest) -> ApiResult<HttpResponse> {
209    let service = policy::http_service_builder().service(service_fn(move |value: HttpRequest| {
210        let client = client.clone();
211        async move { invoke_request(client, value).await }
212    }));
213    service
214        .oneshot(request)
215        .await
216        .map_err(|why| eyre!("HTTP service timeout or middleware error — {why}"))
217}
218/// Utility method to employ best practices when using a Reqwest client to make async HTTP GET requests.
219pub fn get(url: impl Into<String>) -> HttpRequestBuilder {
220    HttpRequestBuilder::new(HttpMethod::Get, url)
221}
222/// Utility method to employ best practices when using a Reqwest client to make async HTTP DELETE requests.
223pub fn delete(url: impl Into<String>) -> HttpRequestBuilder {
224    HttpRequestBuilder::new(HttpMethod::Delete, url)
225}
226/// Utility method to employ best practices when using a Reqwest client to make async HTTP PATCH requests.
227pub fn patch(url: impl Into<String>) -> HttpRequestBuilder {
228    HttpRequestBuilder::new(HttpMethod::Patch, url)
229}
230/// Utility method to employ best practices when using a Reqwest client to make async HTTP POST requests.
231pub fn post(url: impl Into<String>) -> HttpRequestBuilder {
232    HttpRequestBuilder::new(HttpMethod::Post, url)
233}
234/// Utility method to employ best practices when using a Reqwest client to make async HTTP PUT requests.
235pub fn put(url: impl Into<String>) -> HttpRequestBuilder {
236    HttpRequestBuilder::new(HttpMethod::Put, url)
237}
238async fn invoke_request(client: reqwest::Client, request: HttpRequest) -> ApiResult<HttpResponse> {
239    let HttpRequest {
240        headers,
241        json_body,
242        method,
243        url,
244    } = request;
245    let builder = client.request(method.into(), url).header(USER_AGENT, ACORN_USER_AGENT).headers(headers);
246    let builder = match json_body {
247        | Some(value) => builder.json(&value),
248        | None => builder,
249    };
250    match builder.send().await {
251        | Ok(response) => {
252            let status_code = response.status().as_u16();
253            let headers = response.headers().clone();
254            match response.bytes().await {
255                | Ok(body) => Ok(HttpResponse {
256                    body: body.to_vec(),
257                    headers,
258                    status_code,
259                }),
260                | Err(why) => Err(eyre!(why)),
261            }
262        }
263        | Err(why) => Err(eyre!(why)),
264    }
265}
266/// Reads response bytes or converts request, status, and body errors into a contextual error.
267pub async fn response_body_bytes(response: ApiResult<HttpResponse>, error_message: &str) -> ApiResult<Vec<u8>> {
268    match response {
269        | Ok(value) => match value.status_code {
270            | 200..=299 => value.bytes().await.map_err(|why| eyre!("{error_message} — {why}")),
271            | status => Err(eyre!("{error_message} — HTTP {status}")),
272        },
273        | Err(why) => Err(eyre!("{error_message} — {why}")),
274    }
275}
276pub(crate) fn should_retry(method: &HttpMethod, status_code: Option<u16>) -> bool {
277    policy::should_retry(method, status_code)
278}
acorn/io/http/mod.rs

acorn/io/http/
mod.rs
