Transport-level stream buffer primitives shared by the
Nous.HTTP.StreamBackend implementations.
Owns the SSE wire format (parse_sse_buffer/1, parse_sse_event/1),
the 10 MB buffer cap, and the :stream_parser dispatch that lets a
provider swap SSE for another framing (see
Nous.Providers.HTTP.JSONArrayParser).
This module exists so the transport layer (Nous.HTTP.*) does not have
to reach up into the provider layer (Nous.Providers.HTTP) for buffer
handling. Nous.Providers.HTTP keeps thin delegating wrappers for the
public parse_sse_buffer/1 / parse_sse_event/1 API.
Resumable parsers
parse_stream_buffer/3 and flush_stream_buffer/3 thread an opaque
scan_state/0 through the parser. A :stream_parser module MAY
export parse_buffer/2 returning {events, remaining, scan_state} to
resume scanning where the previous chunk left off instead of rescanning
the accumulated buffer from byte 0 — the difference between O(n) and
O(n²) when a single object spans hundreds of chunks. Support is probed
with function_exported?/3; parsers that only export parse_buffer/1
keep working unchanged and simply always get nil back.
Summary
Types
Opaque parser-owned resume token. nil means "no partial scan in
flight" — parse from the start of the buffer.
Functions
Flush the remaining buffer at end of stream.
Resumable form of flush_stream_buffer/2.
Maximum accumulated stream buffer size, in bytes.
Parse an SSE buffer into {events, remaining_buffer}.
Parse a single SSE event into a JSON map, {:stream_done, reason},
{:parse_error, reason}, or nil.
Parse an accumulated stream buffer with the configured parser.
Resumable form of parse_stream_buffer/2.
Types
@type scan_state() :: term() | nil
Opaque parser-owned resume token. nil means "no partial scan in
flight" — parse from the start of the buffer.
Functions
Flush the remaining buffer at end of stream.
SSE needs a trailing \n\n to force the last event through; custom
parsers just re-parse the remaining buffer as-is.
The chunk handler already enforces max_buffer_size/0 on every received
chunk, so the buffer reaching here is by construction within limits. The
synthetic "\n\n" is bookkeeping, not received data — bypass the
public size check so a buffer at exactly the cap doesn't trip a
false-positive overflow on the 2-byte append. Only surface overflow if
the input itself is over.
@spec flush_stream_buffer(String.t(), module() | nil, scan_state()) :: {list(), String.t(), scan_state()}
Resumable form of flush_stream_buffer/2.
@spec max_buffer_size() :: pos_integer()
Maximum accumulated stream buffer size, in bytes.
@spec parse_sse_buffer(String.t() | nil | any()) :: {list(), String.t()} | {:error, :buffer_overflow}
Parse an SSE buffer into {events, remaining_buffer}.
Returns {:error, :buffer_overflow} when the buffer exceeds
max_buffer_size/0. See Nous.Providers.HTTP.parse_sse_buffer/1 for
the documented public entry point.
@spec parse_sse_event(String.t()) :: map() | {:stream_done, String.t()} | {:parse_error, term()} | nil
Parse a single SSE event into a JSON map, {:stream_done, reason},
{:parse_error, reason}, or nil.
See Nous.Providers.HTTP.parse_sse_event/1 for the documented public
entry point.
Parse an accumulated stream buffer with the configured parser.
Non-resumable form; equivalent to parse_stream_buffer(buffer, mod, nil)
with the scan state discarded. Kept for callers that do not thread state.
@spec parse_stream_buffer(String.t(), module() | nil, scan_state()) :: {list(), String.t(), scan_state()}
Resumable form of parse_stream_buffer/2.
Returns {events, remaining_buffer, scan_state}. Pass the returned
scan_state back on the next chunk. Parsers that do not export
parse_buffer/2 always yield nil.
Translates the {:error, :buffer_overflow} tuple from
parse_sse_buffer/1 into the {events, buffer} shape so backends can
stay agnostic about the failure mode.