reqwest_streams/
lib.rs

1#![allow(unused_parens, clippy::new_without_default)]
2#![forbid(unsafe_code)]
3#![cfg_attr(docsrs, feature(doc_cfg))]
4
5//! Streaming responses support for reqwest for different formats:
6//! - JSON array stream format
7//! - JSON Lines (NL/NewLines) format
8//! - CSV stream format
9//! - [Protobuf] len-prefixed stream format
10//! - [Apache Arrow IPC] stream format
11//!
12//! This type of responses are useful when you are reading huge stream of objects from some source (such as database, file, etc)
13//! and want to avoid huge memory allocations to store on the server side.
14//!
15//! # Features
16//!
17//! **Note:** The `default` features do not include any formats.
18//!
19//! - `json`: JSON array and JSON Lines (JSONL) stream formats
20//! - `csv`: CSV stream format
21//! - `protobuf`: [Protobuf] len-prefixed stream format
22//! - `arrow`: [Apache Arrow IPC] stream format
23//! - `tracing`: report progress and errors through [tracing]
24//!
25//! # Example
26//!
27//! ```rust,no_run
28//! use futures::stream::BoxStream as _;
29//! use reqwest_streams::JsonStreamResponse as _;
30//! use serde::Deserialize;
31//!
32//! #[derive(Debug, Clone, Deserialize)]
33//! struct MyTestStructure {
34//!     some_test_field: String
35//! }
36//!
37//!#[tokio::main]
38//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
39//!
40//!     let _stream = reqwest::get("http://localhost:8080/json-array")
41//!         .await?
42//!         .json_array_stream::<MyTestStructure>(1024);
43//!
44//!     Ok(())
45//! }
46//! ```
47//!
48//! More and complete examples available on the github in the examples directory.
49//!
50//! ## Need server support?
51//! There is the same functionality:
52//! - [axum-streams](https://github.com/abdolence/axum-streams-rs).
53//!
54//!
55//! # Observing stream errors
56//!
57//! An error that happens mid-stream is yielded as an item, so a consumer that stops at the
58//! first one silently gets a truncated result. Use [`ReqwestStreamOptions::on_error`] to
59//! observe them, or enable the `tracing` feature to have them logged at `ERROR` on the
60//! `reqwest_streams` target.
61//!
62//! # Observing stream progress
63//!
64//! Nothing at the call site can tell you how much of a response actually arrived, because it
65//! is read long after the call returned. With the `tracing` feature every stream reports its
66//! totals once at `INFO` when it ends, on a `reqwest_streams::response_stream` span:
67//!
68//! ```text
69//! INFO reqwest_streams::response_stream{format="json_array" status=200 items=1000 bytes=28001 elapsed_ms=11239 outcome="completed"}: Finished streaming an HTTP body
70//! ```
71//!
72//! The `outcome` tells apart the three ways a stream can end: `completed`, `aborted` (the
73//! consumer stopped reading early) and `failed`, which reports at `ERROR` instead. Raise the
74//! filter to `reqwest_streams=debug` for a progress line about once a second, and to
75//! `reqwest_streams=trace` for one per body chunk.
76//!
77//! The same accounting is available without tracing, for metrics, via
78//! [`ReqwestStreamOptions::on_progress`].
79//!
80//! [tracing]: https://docs.rs/tracing
81//! [Apache Arrow IPC]: https://arrow.apache.org/docs/format/Columnar.html#serialization-and-interprocess-communication-ipc
82//! [Protobuf]: https://protobuf.dev/programming-guides/encoding/
83
84#[macro_use]
85mod macros;
86
87cfg_json! {
88    pub use json_stream::JsonStreamResponse;
89    mod json_stream;
90    mod json_array_codec;
91}
92
93cfg_csv! {
94    pub use csv_stream::CsvStreamResponse;
95    mod csv_stream;
96}
97
98use crate::error::StreamBodyError;
99
100mod observability;
101pub use observability::{
102    ReqwestStreamErrorHandler, ReqwestStreamOptions, ReqwestStreamOutcome, ReqwestStreamProgress,
103    ReqwestStreamProgressHandler,
104};
105
106cfg_protobuf! {
107    pub use protobuf_stream::ProtobufStreamResponse;
108    mod protobuf_stream;
109    mod protobuf_len_codec;
110}
111
112cfg_arrow! {
113    pub use arrow_ipc_stream::ArrowIpcStreamResponse;
114    mod arrow_ipc_stream;
115    mod arrow_ipc_len_codec;
116}
117
118pub mod error;
119
120/// Alias for the [`Result`] type returned by streaming responses.
121pub type StreamBodyResult<T> = std::result::Result<T, StreamBodyError>;
122
123#[cfg(test)]
124mod test_client;
reqwest_streams/lib.rs

reqwest_streams/
lib.rs
