rustrade_integration/protocol/http/rest/
client.rs

1use crate::{
2    error::SocketError,
3    metric::{Field, Metric, Tag},
4    protocol::http::{BuildStrategy, HttpParser, rest::RestRequest},
5};
6use bytes::Bytes;
7use chrono::Utc;
8use std::borrow::Cow;
9
10/// Configurable REST client capable of executing signed [`RestRequest`]s. Use this when
11/// integrating APIs that require Http in order to interact with resources. Each API will require
12/// a specific combination of [`Signer`](super::super::private::Signer), [`Mac`](hmac::Mac),
13/// signature [`Encoder`](super::super::private::encoder::Encoder), and
14/// [`HttpParser`].
15#[derive(Debug)]
16pub struct RestClient<'a, Strategy, Parser> {
17    /// HTTP [`reqwest::Client`] for executing signed [`reqwest::Request`]s.
18    pub http_client: reqwest::Client,
19
20    /// Base Url of the API being interacted with.
21    pub base_url: Cow<'a, str>,
22
23    /// [`RestRequest`] build strategy for the API being interacted with that implements
24    /// [`BuildStrategy`].
25    ///
26    /// An authenticated [`RestClient`] will utilise API specific
27    /// [`Signer`](super::super::private::Signer) logic, a hashable [`Mac`](hmac::Mac), and a
28    /// signature [`Encoder`](super::super::private::encoder::Encoder). Where as a non authorised
29    /// [`RestRequest`] may add any mandatory `reqwest` headers that are required.
30    pub strategy: Strategy,
31
32    /// [`HttpParser`] that deserialises [`RestRequest::Response`]s, and upon failure parses
33    /// API errors returned from the server.
34    pub parser: Parser,
35}
36
37impl<Strategy, Parser> RestClient<'_, Strategy, Parser>
38where
39    Strategy: BuildStrategy,
40    Parser: HttpParser,
41{
42    /// Execute the provided [`RestRequest`].
43    pub async fn execute<Request>(
44        &self,
45        request: Request,
46    ) -> Result<(Request::Response, Metric), Parser::OutputError>
47    where
48        Request: RestRequest,
49    {
50        // Use provided Request to construct a signed reqwest::Request
51        let request = self.build(request)?;
52
53        // Measure request execution
54        let (status, payload, latency) = self.measured_execution::<Request>(request).await?;
55
56        // Attempt to parse API Success or Error response
57        self.parser
58            .parse::<Request::Response>(status, &payload)
59            .map(|response| (response, latency))
60    }
61
62    /// Use the provided [`RestRequest`] to construct a signed Http [`reqwest::Request`].
63    pub fn build<Request>(&self, request: Request) -> Result<reqwest::Request, SocketError>
64    where
65        Request: RestRequest,
66    {
67        // Construct url
68        let url = format!("{}{}", self.base_url, request.path());
69
70        // Construct RequestBuilder with method & url
71        let mut builder = self
72            .http_client
73            .request(Request::method(), url)
74            .timeout(request.timeout());
75
76        // Add optional query parameters
77        if let Some(query_params) = request.query_params() {
78            builder = builder.query(query_params);
79        }
80
81        // Add optional Body
82        if let Some(body) = request.body() {
83            builder = builder.json(body);
84        }
85
86        // Use RequestBuilder (public or private strategy) to build reqwest::Request
87        self.strategy.build(request, builder)
88    }
89
90    /// Execute the built [`reqwest::Request`] using the [`reqwest::Client`].
91    ///
92    /// Measures and returns the Http request round trip duration.
93    pub async fn measured_execution<Request>(
94        &self,
95        request: reqwest::Request,
96    ) -> Result<(reqwest::StatusCode, Bytes, Metric), SocketError>
97    where
98        Request: RestRequest,
99    {
100        // Construct Http request duration Metric
101        // timestamps are post-1970 (positive)
102        #[allow(clippy::cast_sign_loss)]
103        let time = Utc::now().timestamp_millis() as u64;
104        let mut latency = Metric {
105            name: "http_request_duration",
106            time,
107            tags: vec![
108                Tag::new("http_method", Request::method().as_str()),
109                Tag::new("base_url", self.base_url.as_ref()),
110                Tag::new("path", request.url().path()),
111            ],
112            fields: Vec::with_capacity(1),
113        };
114
115        // Measure the HTTP request round trip duration
116        let start = std::time::Instant::now();
117        let response = self.http_client.execute(request).await?;
118        // as_millis() returns u128; truncation impossible (u64::MAX ms ≈ 584M years)
119        #[allow(clippy::cast_possible_truncation)]
120        let duration = start.elapsed().as_millis() as u64;
121
122        // Update Metric with response status and request duration
123        latency
124            .tags
125            .push(Tag::new("status_code", response.status().as_str()));
126        latency.fields.push(Field::new("duration", duration));
127
128        // Extract Status Code & reqwest::Response Bytes
129        let status_code = response.status();
130        let payload = response.bytes().await?;
131
132        Ok((status_code, payload, latency))
133    }
134}
135
136impl<'a, Strategy, Parser> RestClient<'a, Strategy, Parser> {
137    /// Construct a new [`Self`] using the provided configuration.
138    pub fn new<Url: Into<Cow<'a, str>>>(base_url: Url, strategy: Strategy, parser: Parser) -> Self {
139        Self {
140            http_client: reqwest::Client::new(),
141            base_url: base_url.into(),
142            strategy,
143            parser,
144        }
145    }
146}
147
148#[cfg(test)]
149#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
150mod tests {
151    use super::*;
152    use crate::protocol::http::{HttpParser, public::PublicNoHeaders};
153    use reqwest::StatusCode;
154    use serde::Deserialize;
155    use wiremock::matchers::{method, path};
156    use wiremock::{Mock, MockServer, ResponseTemplate};
157
158    /// `{"value":"decoded"}`, gzip-compressed with a zeroed mtime so the bytes are reproducible:
159    /// `python3 -c 'import gzip; print(gzip.compress(b"{\"value\":\"decoded\"}", mtime=0))'`.
160    const GZIPPED_BODY: &[u8] = &[
161        0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xab, 0x56, 0x2a, 0x4b, 0xcc,
162        0x29, 0x4d, 0x55, 0xb2, 0x52, 0x4a, 0x49, 0x4d, 0xce, 0x4f, 0x49, 0x4d, 0x51, 0xaa, 0x05,
163        0x00, 0xba, 0xf3, 0x5c, 0x77, 0x13, 0x00, 0x00, 0x00,
164    ];
165
166    #[derive(Debug, Deserialize, PartialEq)]
167    struct Payload {
168        value: String,
169    }
170
171    struct GetPayload;
172
173    impl RestRequest for GetPayload {
174        type Response = Payload;
175        type QueryParams = ();
176        type Body = ();
177
178        fn path(&self) -> Cow<'static, str> {
179            Cow::Borrowed("/payload")
180        }
181
182        fn method() -> reqwest::Method {
183            reqwest::Method::GET
184        }
185    }
186
187    struct JsonParser;
188
189    impl HttpParser for JsonParser {
190        type ApiError = serde_json::Value;
191        type OutputError = SocketError;
192
193        fn parse_api_error(&self, status: StatusCode, error: Self::ApiError) -> Self::OutputError {
194            SocketError::HttpResponse(status, error.to_string())
195        }
196    }
197
198    #[tokio::test]
199    async fn the_default_client_negotiates_and_decodes_gzip() {
200        // Pins the workspace's reqwest `gzip` feature: without it the request advertises no
201        // encoding and the compressed body reaches the parser undecoded.
202        let server = MockServer::start().await;
203        Mock::given(method("GET"))
204            .and(path("/payload"))
205            .respond_with(
206                ResponseTemplate::new(200)
207                    .insert_header("content-encoding", "gzip")
208                    .insert_header("content-type", "application/json")
209                    .set_body_bytes(GZIPPED_BODY),
210            )
211            .mount(&server)
212            .await;
213
214        let client = RestClient::new(server.uri(), PublicNoHeaders, JsonParser);
215        let (response, _) = client.execute(GetPayload).await.unwrap();
216
217        assert_eq!(
218            response,
219            Payload {
220                value: "decoded".to_owned()
221            }
222        );
223        let requests = server.received_requests().await.unwrap();
224        let accept_encoding = requests[0]
225            .headers
226            .get("accept-encoding")
227            .and_then(|value| value.to_str().ok());
228        assert!(
229            accept_encoding.is_some_and(|value| value.contains("gzip")),
230            "expected gzip in Accept-Encoding, sent {accept_encoding:?}"
231        );
232    }
233}
rustrade_integration/protocol/http/rest/client.rs

rustrade_integration/protocol/http/rest/
client.rs
