libdd-capabilities 3.0.1

// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/
// SPDX-License-Identifier: Apache-2.0

//! HTTP capability trait and error types.
//!
//! Request and response types are provided by the [`http`] crate, which is a
//! pure-types crate with no platform dependencies (compiles on wasm). The body
//! type is [`bytes::Bytes`].

use crate::maybe_send::{MaybeSend, MaybeSendFuture};
use core::future::Future;
use core::pin::Pin;
use futures_util::StreamExt;

#[derive(Debug, thiserror::Error)]
pub enum HttpError {
    #[error("Network error: {0}")]
    Network(anyhow::Error),
    #[error("Request timed out")]
    Timeout,
    #[error("Response body error: {0}")]
    ResponseBody(anyhow::Error),
    #[error("Invalid request: {0}")]
    InvalidRequest(anyhow::Error),
    #[error("HTTP error: {0}")]
    Other(anyhow::Error),
}

pub type ChunkFuture<'a> = Pin<Box<dyn MaybeSendFuture<Result<(), HttpError>> + 'a>>;

/// A handle for feeding a [`HttpClientCapability::request_streamed`] request body
/// incrementally, one chunk at a time.
pub trait StreamingBodySender: MaybeSend {
    fn send_chunk(&mut self, data: bytes::Bytes) -> ChunkFuture<'_>;
}

/// Fallback [`StreamingBodySender`] that buffers every chunk in memory and only issues
/// the request once the sender side is dropped.
pub struct BufferingBodySender(futures_channel::mpsc::UnboundedSender<bytes::Bytes>);

impl StreamingBodySender for BufferingBodySender {
    fn send_chunk(&mut self, data: bytes::Bytes) -> ChunkFuture<'_> {
        let result = self
            .0
            .unbounded_send(data)
            .map_err(|e| HttpError::Network(e.into()));
        Box::pin(async move { result })
    }
}

pub type ResponseFuture =
    Pin<Box<dyn MaybeSendFuture<Result<http::Response<bytes::Bytes>, HttpError>>>>;

pub type BodySender = Box<dyn StreamingBodySender>;

pub trait HttpClientCapability: Clone + std::fmt::Debug {
    fn new_client() -> Self;

    /// Construct a client for periodic one-shot communication (typically regularly flushing to the
    /// agent or to the backend). Depending on the capabilities of the underlying implementation,
    /// this constructor either:
    ///
    /// - sets the lifetime of pooled connections to a timeout much smaller than 60s (e.g. 5s)
    /// - disables connection pooling entirely if the timeout isn't configurable
    /// - does nothing if there's no connection pooling support to begin with
    ///
    /// The rationale for having limited connection pooling is that we've experienced races when the
    /// connection pooling timeout is higher than the keep-alive timeout of the receiving end. It's
    /// then possible to pick an idle connection and start a request while the connection get
    /// closed at the same time by the receiver, causing an error.
    ///
    /// Connection pooling was initially entirely disabled by this constructor, but it happens that
    /// we send multiple separate requests in a short span of time (e.g. for telemetry on very
    /// short-lived apps). In that situation, if we're agentless, making separate HTTPS connections
    /// is quite costly (can be on the order of magnitude of 0.5sec per connection). Having pooling
    /// with a short lifetime is a better choice, since we can reuse the same connection for those
    /// multiple consecutive requests, while avoiding the race condition.
    fn new_without_connection_pooling() -> Self;

    fn request(
        &self,
        req: http::Request<bytes::Bytes>,
    ) -> impl Future<Output = Result<http::Response<bytes::Bytes>, HttpError>> + MaybeSend;

    /// Like [`Self::request`], but the request body is provided incrementally, one chunk at a
    /// time, via the returned [`BodySender`].
    fn request_streamed(&self, req: http::Request<()>) -> (BodySender, ResponseFuture)
    where
        Self: MaybeSend + 'static,
    {
        let (tx, mut rx) = futures_channel::mpsc::unbounded::<bytes::Bytes>();
        let this = self.clone();
        let fut = async move {
            let mut body = Vec::new();
            while let Some(chunk) = rx.next().await {
                body.extend_from_slice(&chunk);
            }
            this.request(req.map(|()| bytes::Bytes::from(body))).await
        };
        (Box::new(BufferingBodySender(tx)), Box::pin(fut))
    }
}