1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
//! High-performance, `no_std`-compatible utilities for parsing and consuming
//! [Server-Sent Events](https://html.spec.whatwg.org/multipage/server-sent-events.html) (SSE) streams.
//!
//! `sseer` provides a layered API for working with SSE:
//!
//! - [`EventStream`] - a generic [`Stream`][futures_core::Stream] adapter that converts any
//! `Stream<Item = Result<impl AsRef<[u8]>, E>>` into a stream of parsed [`Event`][event::Event]s.
//! - [`EventSource`] (requires `reqwest` feature) - a batteries-included HTTP client that
//! wraps [`reqwest`] with automatic reconnection, retry policies, and the `Last-Event-ID` header.
//! - [`JsonStream`][json_stream::JsonStream] (requires `json` feature) - a stream adapter
//! that deserialises each event's `data` field into a typed value via [`serde_json`].
//! - [`Utf8Stream`][utf8_stream::Utf8Stream] - validates and converts a raw byte stream into
//! a stream of UTF-8 [`Str`][bytes_utils::Str]s, buffering incomplete multi-byte sequences across
//! chunks.
//! - Low-level parsing via [`parser::parse_line`] and [`parser::parse_line_from_buffer`] for
//! custom integrations.
//!
//! # Quick start with `reqwest`
//!
//! ```ignore
//! use futures::StreamExt;
//! use sseer::{EventSource, reqwest::StreamEvent};
//!
//! # #[tokio::main]
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let request = reqwest::Client::new().get("https://example.com/events");
//! let mut source = EventSource::new(request)?;
//!
//! while let Some(result) = source.next().await {
//! match result {
//! Ok(StreamEvent::Open) => println!("connected"),
//! Ok(StreamEvent::Event(evt)) => {
//! println!("{}: {}", evt.event, evt.data);
//! }
//! Err(e) if e.is_response_err() => break,
//! Err(e) => eprintln!("error: {e}"),
//! }
//! }
//! # Ok(())
//! # }
//! ```
//!
//! # Without retry logic
//!
//! If you don't need automatic reconnection, such as with the OpenAI API where you typically
//! can't pick up a dropped stream (or don't want to accidentally send the same request more times), you can skip [`EventSource`]
//! and use [`response_to_stream`] to convert a [`::reqwest::Response`] directly
//! into an [`EventStream`] (the ergonomics are better I recommend it):
//!
//! ```ignore
//! use futures::StreamExt;
//!
//! # #[tokio::main]
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let response = reqwest::Client::new()
//! .get("https://api.example.com/v1/chat/completions")
//! .send()
//! .await?;
//!
//! let mut stream = sseer::response_to_stream(response);
//!
//! while let Some(Ok(event)) = stream.next().await {
//! println!("{}", event.data);
//! }
//! # Ok(())
//! # }
//! ```
//!
//! # Using `EventStream` directly
//!
//! If you already have a byte stream (from any HTTP client, WebSocket, file, etc.)
//! you can use [`EventStream`] without the `reqwest` feature:
//!
//! ```rust
//! use bytes::Bytes;
//! use futures::StreamExt;
//! use sseer::EventStream;
//!
//! # #[tokio::main]
//! # async fn main() {
//! let chunks = vec![
//! Ok::<_, std::io::Error>(Bytes::from("data: hello\n\ndata: world\n\n")),
//! ];
//! let mut stream = EventStream::new(futures::stream::iter(chunks));
//!
//! while let Some(Ok(event)) = stream.next().await {
//! println!("{}", event.data);
//! }
//! # }
//! ```
//!
//! # Feature flags
//!
//! | Feature | Default | Description | no std? |
//! | --- | --- | --- | --- |
//! | `serde` | off | Derives [`Serialize`][::serde::Serialize] and [`Deserialize`][::serde::Deserialize] on [`Event`][event::Event] and enables [`serde`] support in [`bytes-utils`][bytes_utils]. | false |
//! | `std` | off | Enables standard library support in core dependencies (`bytes`, `memchr`, `futures-core`, etc.). Notably enables runtime SIMD for memchr. Turned on automatically by `reqwest` and `json`. | false |
//! | `reqwest` | off | Provides [`EventSource`] for HTTP-based SSE with automatic reconnection and configurable retry policies. | false |
//! | `json` | off | Provides [`JsonStream`][json_stream::JsonStream] for deserialising event data into typed values via [`serde_json`] and lets you choose between the default errors or [`serde_path_to_error`] for richer errors. | false |
//!
//! Without any features enabled, the crate is fully `no_std` compatible and provides
//! [`EventStream`], [`Utf8Stream`][utf8_stream::Utf8Stream], the low-level parser,
//! and retry policy types.
pub
// if the reqwest feature is enabled, this is what someone wants
pub use EventSource;
pub use ;
/// Convert a [`Response`][::reqwest::Response] into a [`Stream`][futures_core::Stream] via a similar mechanism to [::reqwest::Response::bytes_stream]