sse-parse

Incremental parser for Server-Sent Events byte streams. Feed it chunks as they arrive from any transport and drain Events. It implements the WHATWG HTML "interpreting an event stream" algorithm and nothing around it: no HTTP, no reconnection, no MIME checks.

[dependencies]
sse-parse = "0.1"

Two layers

SseParser is the sans-IO core. It works in blocking code, in any async runtime, and on no_std targets with alloc.

use sse_parse::SseParser;

let mut parser = SseParser::new();
parser.feed(b"event: add\ndata: first\n");
parser.feed(b"data: second\n\nid: 7\n");
parser.feed(b"data: third\n\n");

let first = parser.next_event().unwrap();
assert_eq!(first.event, "add");
assert_eq!(first.data, "first\nsecond");

let second = parser.next_event().unwrap();
assert_eq!(second.id, "7");
assert_eq!(parser.last_event_id(), "7");

parser.finish();

EventStream wraps any futures_core::Stream<Item = Result<B, E>> where B: AsRef<[u8]>, which covers Vec<u8>, bytes::Bytes, and HTTP client bodies, and yields Result<Event, Error<E>>. The only error it can yield is Error::Transport(E) from the inner stream. It depends on futures-core alone, so it runs on tokio, async-std, smol, or a hand-rolled executor.

use futures_util::StreamExt;
use sse_parse::EventStream;

let body = reqwest::get("https://example.com/events").await?.bytes_stream();
let mut events = EventStream::new(body);
while let Some(item) = events.next().await {
    let event = item?;
    println!("{}: {}", event.event, event.data);
}

What the tests pin

Every claim below is a test in tests/ that CI runs on stable.

Claim	Test
Exact replay of 36 WHATWG-derived vectors, including the spec's own examples, through the parser and the stream adapter	`conformance.rs`
Chunk boundaries never change the output: every 2-split and 3-split of the short vectors, plus random splits of random bytes, match the single-chunk parse	`chunk_invariance.rs`
Linear time in bytes fed: a 64 KiB line delivered one byte at a time scans each byte at most twice, and T(64 KiB) / T(16 KiB) stays under 8	`perf_gate.rs`
The leading byte order mark is stripped once, at stream offset 0, even one byte per chunk. Later `EF BB BF` bytes are content	`bom_once.rs`
`feed`, `finish`, and `next_event` return on arbitrary bytes and chunkings, and held memory stays proportional to input	`no_panic.rs`, `fuzz/`
Dispatch rules: the last event ID commits at every blank line, including blocks that send no event, and `retry` takes effect when its line is processed	`regressions.rs`

Pinned decisions where the spec is silent:

Invalid UTF-8 becomes U+FFFD and parsing continues. Encoding can never fail, so there is no encoding error to handle.
A carriage return at end of stream terminates its line.
retry values must be all ASCII digits and fit in a u64. Anything else is ignored.
id values containing NUL are ignored and the previous ID stands.

Bounding memory

The default parser holds whatever one block needs. max_buffer(limit) caps that. When a line plus the block's decoded buffers would exceed the limit, the block is dropped and parsing resumes at the next blank line. overflows() counts dropped blocks. The check runs at line granularity, so a capped parser gives the same output for every chunking of the stream.

use sse_parse::SseParser;

let mut parser = SseParser::new().max_buffer(1 << 20);

Where it sits among the SSE crates

Crate	Shape	Pick it when
`sse-parse`	Parser plus a `futures-core` stream adapter. Three dependencies, `no_std` with `alloc`.	You want the WHATWG algorithm with a replay gate, chunking invariance, and linear time, on any runtime.
`eventsource-stream`	Parser plus stream adapter built on `nom`.	You already depend on it. `sse_parse::Event` has the same four fields, so switching is an import change.
`sse-core`	Sans-IO state machine paired with `sse-reqwest-client`. Requires `bytes` and `thiserror`.	You want the matching `reqwest` client with reconnection built in.
`sse-codec`	`futures_codec` encoder and decoder.	You are wiring SSE into a `futures_codec` framed transport, or need to emit events as well as parse them.
`async-sse`	Parser and encoder on `async-std` and `http-types`.	You are on `async-std` with `http-types` already in the tree.

Dependencies and MSRV

memchr for terminator scanning, futures-core for the Stream trait, and pin-project-lite for the adapter. Disable the default std feature for core + alloc. The minimum supported Rust version is 1.65, checked in CI by building the library and running the doctests at that version.

License

MIT. See LICENSE.