libdd_capabilities/
http.rs

1// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/
2// SPDX-License-Identifier: Apache-2.0
3
4//! HTTP capability trait and error types.
5//!
6//! Request and response types are provided by the [`http`] crate, which is a
7//! pure-types crate with no platform dependencies (compiles on wasm). The body
8//! type is [`bytes::Bytes`].
9
10use crate::maybe_send::{MaybeSend, MaybeSendFuture};
11use core::future::Future;
12use core::pin::Pin;
13use futures_util::StreamExt;
14
15#[derive(Debug, thiserror::Error)]
16pub enum HttpError {
17    #[error("Network error: {0}")]
18    Network(anyhow::Error),
19    #[error("Request timed out")]
20    Timeout,
21    #[error("Response body error: {0}")]
22    ResponseBody(anyhow::Error),
23    #[error("Invalid request: {0}")]
24    InvalidRequest(anyhow::Error),
25    #[error("HTTP error: {0}")]
26    Other(anyhow::Error),
27}
28
29pub type ChunkFuture<'a> = Pin<Box<dyn MaybeSendFuture<Result<(), HttpError>> + 'a>>;
30
31/// A handle for feeding a [`HttpClientCapability::request_streamed`] request body
32/// incrementally, one chunk at a time.
33pub trait StreamingBodySender: MaybeSend {
34    fn send_chunk(&mut self, data: bytes::Bytes) -> ChunkFuture<'_>;
35}
36
37/// Fallback [`StreamingBodySender`] that buffers every chunk in memory and only issues
38/// the request once the sender side is dropped.
39pub struct BufferingBodySender(futures_channel::mpsc::UnboundedSender<bytes::Bytes>);
40
41impl StreamingBodySender for BufferingBodySender {
42    fn send_chunk(&mut self, data: bytes::Bytes) -> ChunkFuture<'_> {
43        let result = self
44            .0
45            .unbounded_send(data)
46            .map_err(|e| HttpError::Network(e.into()));
47        Box::pin(async move { result })
48    }
49}
50
51pub type ResponseFuture =
52    Pin<Box<dyn MaybeSendFuture<Result<http::Response<bytes::Bytes>, HttpError>>>>;
53
54pub type BodySender = Box<dyn StreamingBodySender>;
55
56pub trait HttpClientCapability: Clone + std::fmt::Debug {
57    fn new_client() -> Self;
58
59    /// Construct a client for periodic one-shot communication (typically regularly flushing to the
60    /// agent or to the backend). Depending on the capabilities of the underlying implementation,
61    /// this constructor either:
62    ///
63    /// - sets the lifetime of pooled connections to a timeout much smaller than 60s (e.g. 5s)
64    /// - disables connection pooling entirely if the timeout isn't configurable
65    /// - does nothing if there's no connection pooling support to begin with
66    ///
67    /// The rationale for having limited connection pooling is that we've experienced races when the
68    /// connection pooling timeout is higher than the keep-alive timeout of the receiving end. It's
69    /// then possible to pick an idle connection and start a request while the connection get
70    /// closed at the same time by the receiver, causing an error.
71    ///
72    /// Connection pooling was initially entirely disabled by this constructor, but it happens that
73    /// we send multiple separate requests in a short span of time (e.g. for telemetry on very
74    /// short-lived apps). In that situation, if we're agentless, making separate HTTPS connections
75    /// is quite costly (can be on the order of magnitude of 0.5sec per connection). Having pooling
76    /// with a short lifetime is a better choice, since we can reuse the same connection for those
77    /// multiple consecutive requests, while avoiding the race condition.
78    fn new_without_connection_pooling() -> Self;
79
80    fn request(
81        &self,
82        req: http::Request<bytes::Bytes>,
83    ) -> impl Future<Output = Result<http::Response<bytes::Bytes>, HttpError>> + MaybeSend;
84
85    /// Like [`Self::request`], but the request body is provided incrementally, one chunk at a
86    /// time, via the returned [`BodySender`].
87    fn request_streamed(&self, req: http::Request<()>) -> (BodySender, ResponseFuture)
88    where
89        Self: MaybeSend + 'static,
90    {
91        let (tx, mut rx) = futures_channel::mpsc::unbounded::<bytes::Bytes>();
92        let this = self.clone();
93        let fut = async move {
94            let mut body = Vec::new();
95            while let Some(chunk) = rx.next().await {
96                body.extend_from_slice(&chunk);
97            }
98            this.request(req.map(|()| bytes::Bytes::from(body))).await
99        };
100        (Box::new(BufferingBodySender(tx)), Box::pin(fut))
101    }
102}
libdd_capabilities/http.rs

libdd_capabilities/
http.rs
