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 that does not reuse connections.
60    fn new_without_connection_pooling() -> Self;
61
62    fn request(
63        &self,
64        req: http::Request<bytes::Bytes>,
65    ) -> impl Future<Output = Result<http::Response<bytes::Bytes>, HttpError>> + MaybeSend;
66
67    /// Like [`Self::request`], but the request body is provided incrementally, one chunk at a
68    /// time, via the returned [`BodySender`].
69    fn request_streamed(&self, req: http::Request<()>) -> (BodySender, ResponseFuture)
70    where
71        Self: MaybeSend + 'static,
72    {
73        let (tx, mut rx) = futures_channel::mpsc::unbounded::<bytes::Bytes>();
74        let this = self.clone();
75        let fut = async move {
76            let mut body = Vec::new();
77            while let Some(chunk) = rx.next().await {
78                body.extend_from_slice(&chunk);
79            }
80            this.request(req.map(|()| bytes::Bytes::from(body))).await
81        };
82        (Box::new(BufferingBodySender(tx)), Box::pin(fut))
83    }
84}
libdd_capabilities/http.rs

libdd_capabilities/
http.rs
