Pristine.Streaming.SSEDecoder (Pristine v0.4.0)

Stateful decoder for Server-Sent Events (SSE) streams.

Implements the SSE specification: https://html.spec.whatwg.org/multipage/server-sent-events.html

Basic Usage

decoder = SSEDecoder.new()
{events, decoder} = SSEDecoder.feed(decoder, chunk)

Streaming Usage

chunks
|> SSEDecoder.decode_stream()
|> Enum.each(&process_event/1)

SSE Format

The decoder handles the standard SSE format:

event: eventType
id: eventId
retry: 5000
data: line1
data: line2

Events are terminated by a blank line (\n\n, \r\n\r\n, or \r\r). Lines starting with : are comments and are ignored.

Summary

Types

t()

Functions

decode_stream(chunks, opts \\ [])

Create a stream of events from an enumerable of chunks.

feed(decoder, chunk)

Feed a chunk of data to the decoder.

last_event_id(sse_decoder)

Return the last event ID seen by this decoder.

new(opts \\ [])

Create a new decoder with an empty buffer.

Types

t()

@type t() :: %Pristine.Streaming.SSEDecoder{
  buffer: binary(),
  last_event_id: String.t() | nil
}

Functions

decode_stream(chunks, opts \\ [])

@spec decode_stream(Enumerable.t(), keyword()) :: Enumerable.t()

Create a stream of events from an enumerable of chunks.

This is useful for processing SSE responses from HTTP clients that provide chunked data as an enumerable or stream.

Examples

chunks
|> SSEDecoder.decode_stream()
|> Enum.each(fn event ->
  IO.puts("Got event: #{event.data}")
end)

# With JSON parsing
chunks
|> SSEDecoder.decode_stream()
|> Stream.map(&Event.json!/1)
|> Enum.to_list()

feed(decoder, chunk)

@spec feed(t(), binary()) :: {[Pristine.Streaming.Event.t()], t()}

Feed a chunk of data to the decoder.

Returns {events, new_decoder} where events is a list of complete events parsed from the accumulated data.

Examples

iex> decoder = Pristine.Streaming.SSEDecoder.new()
iex> {events, _decoder} = Pristine.Streaming.SSEDecoder.feed(decoder, "data: hello\n\n")
iex> length(events)
1

iex> decoder = Pristine.Streaming.SSEDecoder.new()
iex> {events1, decoder} = Pristine.Streaming.SSEDecoder.feed(decoder, "data: hel")
iex> {events2, _decoder} = Pristine.Streaming.SSEDecoder.feed(decoder, "lo\n\n")
iex> {length(events1), length(events2)}
{0, 1}

last_event_id(sse_decoder)

@spec last_event_id(t()) :: String.t() | nil

Return the last event ID seen by this decoder.

new(opts \\ [])

@spec new(keyword()) :: t()

Create a new decoder with an empty buffer.

Example

decoder = SSEDecoder.new()