qubit_http/request/
http_request_streaming_body.rs

1/*******************************************************************************
2 *
3 *    Copyright (c) 2025 - 2026 Haixing Hu.
4 *
5 *    SPDX-License-Identifier: Apache-2.0
6 *
7 *    Licensed under the Apache License, Version 2.0.
8 *
9 ******************************************************************************/
10//! Deferred streaming request body wrapper.
11
12use std::future::Future;
13use std::pin::Pin;
14use std::sync::Arc;
15
16use super::http_request_body_byte_stream::HttpRequestBodyByteStream;
17
18type HttpRequestBodyStreamFactoryFuture = Pin<Box<dyn Future<Output = HttpRequestBodyByteStream> + Send + 'static>>;
19
20type HttpRequestBodyStreamFactoryFn = dyn Fn() -> HttpRequestBodyStreamFactoryFuture + Send + Sync + 'static;
21
22/// Deferred streaming upload body source.
23///
24/// Each send attempt obtains a fresh byte stream from the stored async factory,
25/// so retries can rebuild the outbound body stream deterministically.
26#[derive(Clone)]
27pub struct HttpRequestStreamingBody {
28    /// Async factory producing one stream for one request attempt.
29    factory: Arc<HttpRequestBodyStreamFactoryFn>,
30}
31
32impl std::fmt::Debug for HttpRequestStreamingBody {
33    /// Formats this type without exposing closure internals.
34    ///
35    /// # Parameters
36    /// - `f`: Formatter destination.
37    ///
38    /// # Returns
39    /// Formatting result.
40    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
41        f.debug_struct("HttpRequestStreamingBody").finish_non_exhaustive()
42    }
43}
44
45impl HttpRequestStreamingBody {
46    /// Creates a deferred streaming body from an async stream factory.
47    ///
48    /// # Parameters
49    /// - `factory`: Async factory that returns a fresh byte stream for one send
50    ///   attempt.
51    ///
52    /// # Returns
53    /// New deferred streaming body wrapper.
54    pub fn new<F>(factory: F) -> Self
55    where
56        F: Fn() -> HttpRequestBodyStreamFactoryFuture + Send + Sync + 'static,
57    {
58        Self {
59            factory: Arc::new(factory),
60        }
61    }
62
63    /// Builds reqwest body from one factory-produced byte stream.
64    ///
65    /// # Returns
66    /// New [`reqwest::Body`] stream instance for one send attempt.
67    pub(crate) async fn to_reqwest_body(&self) -> reqwest::Body {
68        let stream = (self.factory)().await;
69        reqwest::Body::wrap_stream(stream)
70    }
71}
qubit_http/request/http_request_streaming_body.rs

qubit_http/request/
http_request_streaming_body.rs
