libdd_common/
http_common.rs

1// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
2// SPDX-License-Identifier: Apache-2.0
3
4use core::convert::Infallible;
5use core::fmt;
6use core::time::Duration;
7
8use thiserror::Error;
9
10// --- Portable types (available on all platforms including wasm) ---
11
12#[derive(Debug, Clone, Copy)]
13pub enum ErrorKind {
14    Parse,
15    Closed,
16    Canceled,
17    Incomplete,
18    WriteAborted,
19    ParseStatus,
20    Timeout,
21    Other,
22}
23
24#[derive(Debug, Error)]
25pub struct ClientError {
26    source: anyhow::Error,
27    kind: ErrorKind,
28}
29
30impl core::fmt::Display for ClientError {
31    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
32        self.source.fmt(f)
33    }
34}
35
36impl ClientError {
37    pub fn kind(&self) -> ErrorKind {
38        self.kind
39    }
40}
41
42#[derive(Debug)]
43pub enum Error {
44    Client(ClientError),
45    Other(anyhow::Error),
46    Infallible(Infallible),
47}
48
49impl fmt::Display for Error {
50    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
51        match self {
52            Error::Client(e) => write!(f, "client error: {e}"),
53            Error::Infallible(e) => match *e {},
54            Error::Other(e) => write!(f, "other error: {e}"),
55        }
56    }
57}
58
59impl From<std::io::Error> for Error {
60    fn from(value: std::io::Error) -> Self {
61        Self::Other(value.into())
62    }
63}
64
65impl From<http::Error> for Error {
66    fn from(value: http::Error) -> Self {
67        Self::Other(value.into())
68    }
69}
70
71impl core::error::Error for Error {}
72
73// --- Native-only code (hyper, Body, client builders, etc.) ---
74
75#[cfg(not(target_arch = "wasm32"))]
76mod native {
77    use super::*;
78    use core::error::Error as _;
79    use core::task::Poll;
80
81    use crate::connector::Connector;
82    use http_body_util::BodyExt;
83    use hyper::body::Incoming;
84    use hyper_util::rt::TokioTimer;
85    use pin_project::pin_project;
86
87    impl From<hyper::Error> for ClientError {
88        fn from(source: hyper::Error) -> Self {
89            use ErrorKind::*;
90            let kind = if source.is_canceled() {
91                Canceled
92            } else if source.is_parse() {
93                Parse
94            } else if source.is_parse_status() {
95                ParseStatus
96            } else if source.is_incomplete_message() {
97                Incomplete
98            } else if source.is_body_write_aborted() {
99                WriteAborted
100            } else if source.is_timeout() {
101                Timeout
102            } else if source.is_closed() {
103                Closed
104            } else {
105                Other
106            };
107            Self {
108                kind,
109                source: source.into(),
110            }
111        }
112    }
113
114    pub type HttpResponse = http::Response<Body>;
115    pub type HttpRequest = http::Request<Body>;
116    pub type HttpRequestError = hyper_util::client::legacy::Error;
117
118    pub type ResponseFuture = hyper_util::client::legacy::ResponseFuture;
119
120    /// Idle timeout for connections kept alive by a client configured for periodic use (see
121    /// [`new_client_periodic`]).
122    ///
123    /// Kept much smaller than typical keep-alive timeouts on the receiving end (e.g. the Datadog
124    /// agent), so that an idle pooled connection is dropped by our side before the receiver closes
125    /// it.
126    pub(crate) const PERIODIC_POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(5);
127    /// Max number of idle connections in a client's connection pool. This is a safety resource
128    /// bound that we don't really expect to hit in practice.
129    pub(crate) const POOL_MAX_IDLE: usize = 20;
130
131    /// Create a new default configuration hyper client for fixed interval sending.
132    ///
133    /// This client pools connections with a small timeout (smaller than any potential keep-alive on
134    /// the receiver side), because otherwise we would get a pipe closed every second connection
135    /// because of the keep alive in the agent or the backend.
136    ///
137    /// This is in general not a problem if we use the client once every tens of seconds.
138    pub fn new_client_periodic() -> GenericHttpClient<Connector> {
139        hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::default())
140            .pool_timer(TokioTimer::new())
141            .pool_idle_timeout(PERIODIC_POOL_IDLE_TIMEOUT)
142            .pool_max_idle_per_host(POOL_MAX_IDLE)
143            .build(Connector::default())
144    }
145
146    /// Create a new default configuration hyper client.
147    ///
148    /// It will keep connections open for a longer time and reuse them.
149    pub fn new_default_client() -> GenericHttpClient<Connector> {
150        hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::default())
151            .pool_max_idle_per_host(POOL_MAX_IDLE)
152            .build(Connector::default())
153    }
154
155    pub fn into_response(response: hyper::Response<Incoming>) -> HttpResponse {
156        response.map(Body::Incoming)
157    }
158
159    impl From<HttpRequestError> for ClientError {
160        fn from(err: HttpRequestError) -> Self {
161            let kind = if let Some(source) = err.source().and_then(|s| s.downcast_ref::<Error>()) {
162                match source {
163                    Error::Client(client_error) => client_error.kind,
164                    Error::Other(_) => ErrorKind::Other,
165                    Error::Infallible(infallible) => match *infallible {},
166                }
167            } else if err.is_connect() {
168                ErrorKind::Closed
169            } else {
170                ErrorKind::Other
171            };
172            Self {
173                source: err.into(),
174                kind,
175            }
176        }
177    }
178
179    pub async fn collect_response_bytes(response: HttpResponse) -> Result<bytes::Bytes, Error> {
180        Ok(response.into_body().collect().await?.to_bytes())
181    }
182
183    pub fn mock_response(
184        builder: http::response::Builder,
185        body: hyper::body::Bytes,
186    ) -> anyhow::Result<HttpResponse> {
187        Ok(builder.body(Body::from_bytes(body))?)
188    }
189
190    pub fn empty_response(builder: http::response::Builder) -> Result<HttpResponse, Error> {
191        Ok(builder.body(Body::empty())?)
192    }
193
194    #[pin_project(project=BodyProj)]
195    #[derive(Debug)]
196    pub enum Body {
197        Single(#[pin] http_body_util::Full<hyper::body::Bytes>),
198        Empty(#[pin] http_body_util::Empty<hyper::body::Bytes>),
199        Boxed(#[pin] http_body_util::combinators::BoxBody<hyper::body::Bytes, anyhow::Error>),
200        Channel(#[pin] tokio::sync::mpsc::Receiver<hyper::body::Bytes>),
201        Incoming(#[pin] hyper::body::Incoming),
202    }
203
204    pub struct Sender {
205        tx: tokio::sync::mpsc::Sender<hyper::body::Bytes>,
206    }
207
208    impl Sender {
209        pub async fn send_data(&self, data: hyper::body::Bytes) -> anyhow::Result<()> {
210            self.tx.send(data).await?;
211            Ok(())
212        }
213    }
214
215    impl Body {
216        pub fn empty() -> Self {
217            Body::Empty(http_body_util::Empty::new())
218        }
219
220        pub fn from_bytes(bytes: hyper::body::Bytes) -> Self {
221            Body::Single(http_body_util::Full::new(bytes))
222        }
223
224        pub fn boxed<
225            E: core::error::Error + Sync + Send + 'static,
226            T: hyper::body::Body<Data = hyper::body::Bytes, Error = E> + Sync + Send + 'static,
227        >(
228            body: T,
229        ) -> Self {
230            Body::Boxed(body.map_err(anyhow::Error::from).boxed())
231        }
232
233        pub fn channel() -> (Sender, Self) {
234            let (tx, rx) = tokio::sync::mpsc::channel(1);
235            (Sender { tx }, Body::Channel(rx))
236        }
237
238        pub fn incoming(incoming: Incoming) -> Self {
239            Body::Incoming(incoming)
240        }
241    }
242
243    impl Default for Body {
244        fn default() -> Self {
245            Body::empty()
246        }
247    }
248
249    impl From<&'static str> for Body {
250        fn from(s: &'static str) -> Self {
251            Body::from_bytes(hyper::body::Bytes::from_static(s.as_bytes()))
252        }
253    }
254
255    impl From<Vec<u8>> for Body {
256        fn from(s: Vec<u8>) -> Self {
257            Body::from_bytes(hyper::body::Bytes::from(s))
258        }
259    }
260
261    impl From<String> for Body {
262        fn from(s: String) -> Self {
263            Body::from_bytes(hyper::body::Bytes::from(s))
264        }
265    }
266
267    impl hyper::body::Body for Body {
268        type Data = hyper::body::Bytes;
269
270        type Error = Error;
271
272        fn poll_frame(
273            self: core::pin::Pin<&mut Self>,
274            cx: &mut core::task::Context<'_>,
275        ) -> core::task::Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
276            match self.project() {
277                BodyProj::Single(pin) => pin.poll_frame(cx).map_err(Error::Infallible),
278                BodyProj::Empty(pin) => pin.poll_frame(cx).map_err(Error::Infallible),
279                BodyProj::Boxed(pin) => pin.poll_frame(cx).map_err(Error::Other),
280                BodyProj::Channel(pin) => {
281                    let data = match pin.get_mut().poll_recv(cx) {
282                        Poll::Ready(Some(data)) => data,
283                        Poll::Ready(None) => return Poll::Ready(None),
284                        Poll::Pending => return Poll::Pending,
285                    };
286                    Poll::Ready(Some(Ok(hyper::body::Frame::data(data))))
287                }
288                BodyProj::Incoming(pin) => pin
289                    .poll_frame(cx)
290                    .map_err(|e| Error::Client(ClientError::from(e))),
291            }
292        }
293
294        fn is_end_stream(&self) -> bool {
295            match self {
296                Body::Single(body) => body.is_end_stream(),
297                Body::Empty(body) => body.is_end_stream(),
298                Body::Boxed(body) => body.is_end_stream(),
299                Body::Channel(body) => body.is_closed() && body.is_empty(),
300                Body::Incoming(body) => body.is_end_stream(),
301            }
302        }
303
304        fn size_hint(&self) -> http_body::SizeHint {
305            match self {
306                Body::Single(body) => body.size_hint(),
307                Body::Empty(body) => body.size_hint(),
308                Body::Boxed(body) => body.size_hint(),
309                Body::Channel(_) => http_body::SizeHint::default(),
310                Body::Incoming(body) => body.size_hint(),
311            }
312        }
313    }
314
315    pub type GenericHttpClient<C> = hyper_util::client::legacy::Client<C, Body>;
316
317    pub fn client_builder() -> hyper_util::client::legacy::Builder {
318        hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::default())
319    }
320}
321
322#[cfg(not(target_arch = "wasm32"))]
323pub use native::*;
libdd_common/http_common.rs

libdd_common/
http_common.rs
