slog_cloud/
client.rs

1use std::{
2    fmt::{self, Debug, Display, Formatter},
3    time::Duration,
4};
5
6use bytes::Bytes;
7use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
8
9pub use reqwest::{Method, StatusCode, Url, header};
10
11use crate::error::Error;
12
13/// Log message send error.
14pub struct SendError {
15    inner: Error,
16    can_retry: bool,
17}
18
19impl SendError {
20    /// Check if the send request can be retried.
21    #[inline]
22    pub fn can_retry(&self) -> bool {
23        self.can_retry
24    }
25}
26
27impl Debug for SendError {
28    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
29        f.debug_struct("SendError")
30            .field("inner", &self.inner)
31            .finish_non_exhaustive()
32    }
33}
34
35impl Display for SendError {
36    #[inline]
37    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
38        Display::fmt(&self.inner, f)
39    }
40}
41
42impl std::error::Error for SendError {
43    #[inline]
44    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
45        Some(&self.inner)
46    }
47}
48
49impl From<SendError> for Error {
50    #[inline]
51    fn from(err: SendError) -> Self {
52        err.inner
53    }
54}
55
56/// Cloud service client.
57#[trait_variant::make(Send)]
58pub trait Client {
59    type Message;
60
61    /// Send a given log message to the cloud service.
62    async fn send(&self, msg: Self::Message) -> Result<(), SendError>;
63}
64
65/// HTTP client error.
66#[derive(Debug)]
67pub enum HttpClientError {
68    UnexpectedStatusCode(StatusCode, Bytes),
69    Other(Error),
70}
71
72impl Display for HttpClientError {
73    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
74        match self {
75            Self::UnexpectedStatusCode(status, body) => {
76                write!(f, "server responded with HTTP {status}")?;
77
78                if f.alternate() {
79                    write!(f, ":\n{}", String::from_utf8_lossy(body.as_ref()))?;
80                }
81
82                Ok(())
83            }
84            Self::Other(err) => Display::fmt(err, f),
85        }
86    }
87}
88
89impl std::error::Error for HttpClientError {
90    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
91        match self {
92            Self::UnexpectedStatusCode(_, _) => None,
93            Self::Other(err) => Some(err),
94        }
95    }
96}
97
98impl From<Error> for HttpClientError {
99    #[inline]
100    fn from(err: Error) -> Self {
101        Self::Other(err)
102    }
103}
104
105impl From<HttpClientError> for Error {
106    fn from(err: HttpClientError) -> Self {
107        match err {
108            HttpClientError::Other(err) => err,
109            err => Error::from_cause(err),
110        }
111    }
112}
113
114impl From<HttpClientError> for SendError {
115    fn from(err: HttpClientError) -> Self {
116        let can_retry = match err {
117            HttpClientError::UnexpectedStatusCode(status, _) => {
118                let code = status.as_u16();
119
120                code == 408 || code == 429 || code >= 500
121            }
122            HttpClientError::Other(_) => true,
123        };
124
125        Self {
126            inner: err.into(),
127            can_retry,
128        }
129    }
130}
131
132/// HTTP client builder.
133pub struct HttpClientBuilder {
134    method: Method,
135    headers: HeaderMap<HeaderValue>,
136    request_timeout: Duration,
137}
138
139impl HttpClientBuilder {
140    /// Create a new builder.
141    fn new() -> Self {
142        Self {
143            method: Method::POST,
144            headers: HeaderMap::new(),
145            request_timeout: Duration::from_secs(60),
146        }
147    }
148
149    /// Set the request method.
150    #[inline]
151    pub fn method(mut self, method: Method) -> Self {
152        self.method = method;
153        self
154    }
155
156    /// Set a give request header.
157    pub fn header<K, V>(mut self, key: K, value: V) -> Result<Self, Error>
158    where
159        HeaderName: TryFrom<K>,
160        <HeaderName as TryFrom<K>>::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
161        HeaderValue: TryFrom<V>,
162        <HeaderValue as TryFrom<V>>::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
163    {
164        let key: HeaderName = key
165            .try_into()
166            .map_err(|err| Error::from_static_msg_and_cause("invalid header name", err))?;
167
168        let value: HeaderValue = value
169            .try_into()
170            .map_err(|err| Error::from_static_msg_and_cause("invalid header value", err))?;
171
172        self.headers.insert(key, value);
173
174        Ok(self)
175    }
176
177    /// Set the request headers.
178    #[inline]
179    pub fn headers(mut self, headers: HeaderMap<HeaderValue>) -> Self {
180        self.headers = headers;
181        self
182    }
183
184    /// Set the request timeout (the default is 60 seconds).
185    #[inline]
186    pub fn request_timeout(mut self, timeout: Duration) -> Self {
187        self.request_timeout = timeout;
188        self
189    }
190
191    /// Create the HTTP client.
192    pub fn build(self, url: Url) -> Result<HttpClient, Error> {
193        let client = reqwest::Client::builder()
194            .timeout(self.request_timeout)
195            .build()
196            .map_err(|err| {
197                Error::from_static_msg_and_cause("unable to create an HTTP client", err)
198            })?;
199
200        let res = HttpClient {
201            url,
202            method: self.method,
203            headers: self.headers,
204            client,
205        };
206
207        Ok(res)
208    }
209}
210
211/// HTTP client.
212pub struct HttpClient {
213    url: Url,
214    method: Method,
215    headers: HeaderMap<HeaderValue>,
216    client: reqwest::Client,
217}
218
219impl HttpClient {
220    /// Get an HTTP client builder.
221    #[inline]
222    pub fn builder() -> HttpClientBuilder {
223        HttpClientBuilder::new()
224    }
225
226    /// Send a given message.
227    pub async fn send(&self, body: Bytes) -> Result<(), HttpClientError> {
228        let method = self.method.clone();
229        let url = self.url.clone();
230        let headers = self.headers.clone();
231
232        let response = self
233            .client
234            .request(method, url)
235            .headers(headers)
236            .body(body)
237            .send()
238            .await
239            .map_err(|err| Error::from_static_msg_and_cause("unable to send a request", err))?;
240
241        let status = response.status();
242
243        let body = response.bytes().await.map_err(|err| {
244            Error::from_static_msg_and_cause("unable to read a response body", err)
245        })?;
246
247        if status.is_success() {
248            Ok(())
249        } else {
250            Err(HttpClientError::UnexpectedStatusCode(status, body))
251        }
252    }
253}
254
255#[cfg(test)]
256mod tests {
257    use bytes::Bytes;
258    use tokio::{
259        io::{AsyncReadExt, AsyncWriteExt},
260        net::{TcpListener, TcpStream},
261        task::JoinHandle,
262    };
263
264    use crate::error::Error;
265
266    use super::{HttpClient, HttpClientError, Method, SendError, StatusCode, Url};
267
268    /// Accept a single HTTP request and respond with a given status code and
269    /// body.
270    ///
271    /// The returned task resolves to the request head and body.
272    async fn serve_once(status: u16, body: &'static str) -> (Url, JoinHandle<(String, Vec<u8>)>) {
273        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
274
275        let local_addr = listener.local_addr();
276
277        let url = format!("http://{}/some/path", local_addr.unwrap())
278            .parse()
279            .unwrap();
280
281        let server = tokio::spawn(async move {
282            let (mut stream, _) = listener.accept().await.unwrap();
283
284            let mut request = Vec::new();
285
286            let head_len = loop {
287                read_more(&mut stream, &mut request).await;
288
289                if let Some(pos) = request.windows(4).position(|w| w == b"\r\n\r\n") {
290                    break pos + 4;
291                }
292            };
293
294            let head = std::str::from_utf8(&request[..head_len])
295                .unwrap()
296                .to_string();
297
298            let content_length = head
299                .lines()
300                .filter_map(|line| line.split_once(':'))
301                .find(|(name, _)| name.eq_ignore_ascii_case("content-length"))
302                .map(|(_, value)| value.trim().parse().unwrap())
303                .unwrap_or(0);
304
305            while request.len() < head_len + content_length {
306                read_more(&mut stream, &mut request).await;
307            }
308
309            let response = format!(
310                "HTTP/1.1 {status} Status\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
311                body.len()
312            );
313
314            stream.write_all(response.as_bytes()).await.unwrap();
315
316            (head, request.split_off(head_len))
317        });
318
319        (url, server)
320    }
321
322    /// Read more data from a given stream into a given buffer.
323    async fn read_more(stream: &mut TcpStream, buffer: &mut Vec<u8>) {
324        let mut chunk = [0u8; 4096];
325
326        let len = stream.read(&mut chunk).await.unwrap();
327
328        assert!(len > 0, "connection closed");
329
330        buffer.extend_from_slice(&chunk[..len]);
331    }
332
333    #[test]
334    fn only_transient_errors_can_be_retried() {
335        let cases = [
336            (400, false),
337            (401, false),
338            (403, false),
339            (404, false),
340            (408, true),
341            (413, false),
342            (429, true),
343            (500, true),
344            (502, true),
345            (503, true),
346        ];
347
348        for (code, expected) in cases {
349            let status = StatusCode::from_u16(code).unwrap();
350
351            let err = SendError::from(HttpClientError::UnexpectedStatusCode(status, Bytes::new()));
352
353            assert_eq!(err.can_retry(), expected, "HTTP {code}");
354        }
355
356        let err = SendError::from(HttpClientError::Other(Error::from_static_msg(
357            "connection reset",
358        )));
359
360        assert!(err.can_retry());
361    }
362
363    #[tokio::test]
364    async fn http_client_sends_configured_request() {
365        let (url, server) = serve_once(200, "").await;
366
367        let client = HttpClient::builder()
368            .method(Method::PUT)
369            .header("X-Test", "value")
370            .unwrap()
371            .build(url)
372            .unwrap();
373
374        client.send(Bytes::from_static(b"payload")).await.unwrap();
375
376        let (head, body) = server.await.unwrap();
377
378        assert!(head.starts_with("PUT /some/path HTTP/1.1\r\n"), "{head}");
379        assert!(
380            head.to_ascii_lowercase().contains("\r\nx-test: value\r\n"),
381            "{head}"
382        );
383        assert_eq!(body, b"payload");
384    }
385
386    #[tokio::test]
387    async fn http_client_reports_unexpected_status() {
388        let (url, server) = serve_once(503, "busy").await;
389
390        let client = HttpClient::builder().build(url).unwrap();
391
392        match client.send(Bytes::new()).await {
393            Err(HttpClientError::UnexpectedStatusCode(status, body)) => {
394                assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
395                assert_eq!(body, "busy");
396            }
397            res => panic!("unexpected result: {res:?}"),
398        }
399
400        server.await.unwrap();
401    }
402}
slog_cloud/client.rs

slog_cloud/
client.rs
