cloud_sdk_reqwest/asynchronous/
client.rs

1use core::fmt;
2
3use cloud_sdk::Method;
4use cloud_sdk::transport::{AsyncTransport, StatusCode, TransportRequest, TransportResponse};
5use cloud_sdk_sanitization::sanitize_bytes;
6use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, HeaderValue};
7use reqwest::{Body, Client};
8
9use crate::shared::{BearerToken, HttpsEndpoint, TransportError, parse_rate_limit};
10
11use super::body::SanitizedBuffer;
12
13/// Hardened provider-neutral reqwest asynchronous transport.
14///
15/// The adapter uses reqwest's Tokio-based execution internally but does not
16/// install or own a runtime. Callers must poll it from a compatible executor.
17pub struct AsyncClient {
18    client: Client,
19    endpoint: HttpsEndpoint,
20    token: BearerToken,
21}
22
23impl AsyncClient {
24    pub(super) const fn new(client: Client, endpoint: HttpsEndpoint, token: BearerToken) -> Self {
25        Self {
26            client,
27            endpoint,
28            token,
29        }
30    }
31
32    async fn send_inner<'buffer>(
33        &mut self,
34        request: TransportRequest<'_>,
35        response_body: &'buffer mut [u8],
36    ) -> Result<TransportResponse<'buffer>, TransportError> {
37        sanitize_bytes(response_body);
38        let url = self
39            .endpoint
40            .compose(request.target())
41            .map_err(|_| TransportError::TargetRejected)?;
42        let authorization = self
43            .token
44            .header_value()
45            .map_err(|_| TransportError::HeaderRejected)?;
46        let mut outbound = self
47            .client
48            .request(map_method(request.method()), url)
49            .header(AUTHORIZATION, authorization)
50            .header(ACCEPT, HeaderValue::from_static("application/json"));
51
52        if let Some(content_type) = request.content_type() {
53            let value = HeaderValue::from_str(content_type.as_str())
54                .map_err(|_| TransportError::HeaderRejected)?;
55            outbound = outbound.header(CONTENT_TYPE, value);
56        } else if !request.body().is_empty() {
57            return Err(TransportError::MissingContentType);
58        }
59
60        if !request.body().is_empty() {
61            let body = SanitizedBuffer::copy_from(request.body())
62                .map_err(|_| TransportError::RequestBodyAllocationFailed)?;
63            let _ = u64::try_from(request.body().len())
64                .map_err(|_| TransportError::RequestBodyTooLarge)?;
65            outbound = outbound.body(Body::from(body.into_bytes()));
66        }
67
68        let mut response = outbound.send().await.map_err(classify_reqwest_error)?;
69        self.endpoint
70            .verify_origin(response.url())
71            .map_err(|_| TransportError::ResponseOriginChanged)?;
72        if response.content_length().is_some_and(|length| {
73            u64::try_from(response_body.len()).map_or(true, |cap| length > cap)
74        }) {
75            return Err(TransportError::ResponseTooLarge);
76        }
77        let status =
78            StatusCode::new(response.status().as_u16()).ok_or(TransportError::InvalidStatus)?;
79        let rate_limit = parse_rate_limit(response.headers())?;
80        let buffered = read_response(&mut response, response_body.len()).await?;
81        let body_len = buffered.len();
82        let initialized = response_body
83            .get_mut(..body_len)
84            .ok_or(TransportError::ResponseReadFailed)?;
85        initialized.copy_from_slice(buffered.as_ref());
86        let response = TransportResponse::new(status, initialized);
87        Ok(rate_limit.map_or(response, |value| response.with_rate_limit(value)))
88    }
89}
90
91impl AsyncTransport for AsyncClient {
92    type Error = TransportError;
93
94    async fn send<'transport, 'request, 'buffer>(
95        &'transport mut self,
96        request: TransportRequest<'request>,
97        response_body: &'buffer mut [u8],
98    ) -> Result<TransportResponse<'buffer>, Self::Error>
99    where
100        'request: 'transport,
101        'buffer: 'transport,
102    {
103        self.send_inner(request, response_body).await
104    }
105}
106
107impl fmt::Debug for AsyncClient {
108    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
109        formatter
110            .debug_struct("AsyncClient")
111            .field("endpoint", &"[redacted]")
112            .field("token", &"[redacted]")
113            .finish_non_exhaustive()
114    }
115}
116
117async fn read_response(
118    response: &mut reqwest::Response,
119    limit: usize,
120) -> Result<SanitizedBuffer, TransportError> {
121    let mut buffered = SanitizedBuffer::with_capacity(limit)
122        .map_err(|_| TransportError::ResponseBodyAllocationFailed)?;
123    loop {
124        let chunk = response
125            .chunk()
126            .await
127            .map_err(|_| TransportError::ResponseReadFailed)?;
128        let Some(chunk) = chunk else { break };
129        buffered
130            .extend_bounded(&chunk, limit)
131            .map_err(|_| TransportError::ResponseTooLarge)?;
132    }
133    Ok(buffered)
134}
135
136const fn map_method(method: Method) -> reqwest::Method {
137    match method {
138        Method::Get => reqwest::Method::GET,
139        Method::Post => reqwest::Method::POST,
140        Method::Put => reqwest::Method::PUT,
141        Method::Delete => reqwest::Method::DELETE,
142    }
143}
144
145fn classify_reqwest_error(error: reqwest::Error) -> TransportError {
146    if error.is_timeout() {
147        TransportError::TimedOut
148    } else if error.is_connect() {
149        TransportError::ConnectFailed
150    } else {
151        TransportError::RequestFailed
152    }
153}
cloud_sdk_reqwest/asynchronous/client.rs

cloud_sdk_reqwest/asynchronous/
client.rs
