reqwest_streams/
json_body.rs

1//! Streaming a JSON request body.
2
3use crate::stream_body::{ReqwestStreamBody, StreamBodyRequest};
4use futures::Stream;
5use http_streams_core::{JsonArrayStreamFormat, JsonNewLineStreamFormat};
6use serde::Serialize;
7
8/// Extension trait for [`reqwest::RequestBuilder`] that streams a JSON request body.
9///
10/// The `try_` variants take a fallible source stream. See [`ReqwestStreamBody`] for the HTTP
11/// caveats that apply to every streamed request body — the redirect one in particular.
12pub trait JsonStreamRequest {
13    /// Streams `stream` as a JSON array, setting `Content-Type: application/json`.
14    ///
15    /// ```rust,no_run
16    /// use futures::stream;
17    /// use reqwest_streams::JsonStreamRequest as _;
18    /// use serde::Serialize;
19    ///
20    /// #[derive(Serialize)]
21    /// struct MyItem {
22    ///     field: String,
23    /// }
24    ///
25    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
26    /// let items = stream::iter(vec![MyItem { field: "value".into() }]);
27    ///
28    /// reqwest::Client::new()
29    ///     .post("http://localhost:8080/ingest")
30    ///     .json_array_stream_body(items)
31    ///     .send()
32    ///     .await?;
33    /// # Ok(())
34    /// # }
35    /// ```
36    fn json_array_stream_body<S, T>(self, stream: S) -> reqwest::RequestBuilder
37    where
38        T: Serialize + Send + 'static,
39        S: Stream<Item = T> + Send + 'static;
40
41    /// Streams a fallible `stream` as a JSON array.
42    fn try_json_array_stream_body<S, T, E>(self, stream: S) -> reqwest::RequestBuilder
43    where
44        T: Serialize + Send + 'static,
45        S: Stream<Item = Result<T, E>> + Send + 'static,
46        E: Into<Box<dyn std::error::Error + Send + Sync>> + Send + 'static;
47
48    /// Streams `stream` as JSON Lines, setting `Content-Type: application/jsonstream`.
49    fn json_nl_stream_body<S, T>(self, stream: S) -> reqwest::RequestBuilder
50    where
51        T: Serialize + Send + 'static,
52        S: Stream<Item = T> + Send + 'static;
53
54    /// Streams a fallible `stream` as JSON Lines.
55    fn try_json_nl_stream_body<S, T, E>(self, stream: S) -> reqwest::RequestBuilder
56    where
57        T: Serialize + Send + 'static,
58        S: Stream<Item = Result<T, E>> + Send + 'static,
59        E: Into<Box<dyn std::error::Error + Send + Sync>> + Send + 'static;
60}
61
62impl JsonStreamRequest for reqwest::RequestBuilder {
63    fn json_array_stream_body<S, T>(self, stream: S) -> reqwest::RequestBuilder
64    where
65        T: Serialize + Send + 'static,
66        S: Stream<Item = T> + Send + 'static,
67    {
68        self.stream_body(ReqwestStreamBody::new(JsonArrayStreamFormat::new(), stream))
69    }
70
71    fn try_json_array_stream_body<S, T, E>(self, stream: S) -> reqwest::RequestBuilder
72    where
73        T: Serialize + Send + 'static,
74        S: Stream<Item = Result<T, E>> + Send + 'static,
75        E: Into<Box<dyn std::error::Error + Send + Sync>> + Send + 'static,
76    {
77        self.stream_body(ReqwestStreamBody::try_new(
78            JsonArrayStreamFormat::new(),
79            stream,
80        ))
81    }
82
83    fn json_nl_stream_body<S, T>(self, stream: S) -> reqwest::RequestBuilder
84    where
85        T: Serialize + Send + 'static,
86        S: Stream<Item = T> + Send + 'static,
87    {
88        self.stream_body(ReqwestStreamBody::new(
89            JsonNewLineStreamFormat::new(),
90            stream,
91        ))
92    }
93
94    fn try_json_nl_stream_body<S, T, E>(self, stream: S) -> reqwest::RequestBuilder
95    where
96        T: Serialize + Send + 'static,
97        S: Stream<Item = Result<T, E>> + Send + 'static,
98        E: Into<Box<dyn std::error::Error + Send + Sync>> + Send + 'static,
99    {
100        self.stream_body(ReqwestStreamBody::try_new(
101            JsonNewLineStreamFormat::new(),
102            stream,
103        ))
104    }
105}
reqwest_streams/json_body.rs

reqwest_streams/
json_body.rs
