Skip to content
apomesh

apomesh-substrate-sse

The generic line-oriented SSE parser — the substrate’s one decoder for streamed text/event-stream bodies. It is the sibling of apomesh-substrate-net: that leaf owns how the substrate reaches an HTTP endpoint, this one owns how a streamed SSE body is decoded. It was extracted from the LLM provider plane’s private parser when the outbound A2A client became the second consumer.

Deliberately hand-rolled and dependency-light — tokio io + channel primitives only, no external SSE crate — so the parser is cheap to audit, sandbox-friendly, and shared as a substrate primitive.

Position in the workspace

Section titled “Position in the workspace”
  • Layer: support — a workspace utility leaf beside the layering, like apomesh-substrate-net.
  • Depends on: no workspace crates — tokio (io-util, rt, sync), tokio-stream, thiserror only.
  • Consumed by: apomesh-substrate-llm (the streaming LLM provider clients) and apomesh-a2a-client (the outbound A2A streaming path).

What it owns

Section titled “What it owns”

The WHATWG SSE subset (HTML Living Standard §9.2) the substrate’s streaming consumers need:

  • data: lines (one or more per event, concatenated with \n per the spec) and optional event: type headers — Anthropic emits event names, OpenAI does not, and A2A streams JSON-RPC results; all three decode through the same parser.
  • Empty-line event termination with both CRLF and bare-LF accepted; id:, retry:, and : comment lines are recognised and skipped (no current consumer uses them).
  • EOF flushes a buffered, unterminated event, so a final chunk is not lost when a server closes without a trailing blank line.
  • OpenAI’s data: [DONE] terminator sentinel surfaces as a regular event — the per-consumer translator decides what it means.

Fragmented frames across TCP reads, leading-byte splits, and UTF-8 boundary issues are absorbed by read_line, which only yields complete lines.

Public surface

Section titled “Public surface”

Verified against lib.rs:

  • SseEvent { event: Option<String>, data: String } — one decoded event. Empty events (no data: and no event:) are absorbed, never emitted.
  • SseParseError — the typed read-side failure; transport errors surface as strings so reqwest-specific types stay out of the substrate’s public surface.
  • parse_sse_stream(reader) -> impl Stream<Item = Result<SseEvent, SseParseError>> — the one entry point: feed it any AsyncBufRead body, consume a stream of events.
Section titled “Related”