🥮 mooncakes.io

    #gaato/mttp/sse

    Server-sent events (text/event-stream) as the WHATWG HTML standard defines them: an incremental parser, a shared async reader over the portable @http.BodyStream interface, and an encoder for the server side. No runtime adapter is required.

    • SseParser::feed takes chunks however the transport splits them, even in the middle of a UTF-8 character or a CRLF, and returns the events they complete. Lines end at CR, LF or CRLF; a byte order mark is skipped at the start of the stream only; malformed UTF-8 becomes U+FFFD; comments and unknown fields are ignored; an id containing NUL is ignored.
    • SseParser::finish ends the stream and drops an event that never got its blank line, as the standard says.
    • last_event_id and reconnection_ms are the stream's state for the next connection: what to send as Last-Event-ID and how long the server asked the client to wait. Waits past one hour read as one hour.
    • encode writes one event, refusing an event name or id that would inject another field.

    The parser is bounded: one event, counting its data lines, the event and id lines in effect and the line being read, may hold max_event_bytes (16 MiB by default). Past that, feed raises @http.TransportError(TooLarge(limit), Body) and the stream ends; events that the same chunk completed earlier are returned first and the next call raises.

    SseReader::new(source, max_event_bytes~) owns a body source and exposes next_event for pull consumers and each(receive, stop~) for callbacks. It delivers queued events before surfacing a deferred overflow, and checks that overflow before reading again so a quiet peer cannot hide it. Each event has the parser's byte limit; the queue holds at most the events from one source chunk. Both client.as_sse and endpoint SSE outputs use this reader.

    EOF, read/parse failure and cancellation during a pull close the source. each also closes on callback failure or cancellation and stops before delivering a matching sentinel. An intentional stop discards any queued events and pending overflow. Pull consumers stopping early must call close; later pulls return EOF without reading the source. As with a body stream, use one consumer without concurrent pulls.

    ///|
    test "parse an event stream split at arbitrary points" {
      let parser = @sse.SseParser::new()
      let events = parser.feed(b"retry: 1000\nid: 1\nevent: delta\ndata: hel")
      assert_eq(events, [])
      let events = parser.feed(b"lo\r\n\r\ndata: [DONE]\n\n")
      assert_eq(events.map(e => (e.event, e.data)), [
        ("delta", "hello"),
        ("message", "[DONE]"),
      ])
      assert_eq(parser.finish(), [])
      assert_eq(parser.last_event_id(), Some("1"))
      assert_eq(parser.reconnection_ms(), Some(1000))
    }

    ///|
    test "encode an event" {
      let event : @sse.SseEvent = {
        event: "delta",
        data: "two\nlines",
        id: Some("2"),
        retry: None,
      }
      assert_eq(
        @sse.encode(event),
        b"event: delta\nid: 2\ndata: two\ndata: lines\n\n",
      )
    }

    SseEvent

    source
    pub(all) struct SseEvent {
      event : String
      data : String
      id : String?
      retry : Int?
    } derive(Eq, 
    Debug    )

    A dispatched event, with resolved type and the current last-event-id. Retry is the last valid retry seen since the previous dispatched event, in milliseconds, capped at one hour.

    SseEvent::equal

    source
    fn SseEvent::equal(SseEvent, SseEvent) -> Bool

    Compares all event fields.

    SseEvent::new

    source
    fn SseEvent::new(data : String, event? : String, id? : String, retry? : Int) -> SseEvent

    Creates an event of type event (by default message, the type a stream gives an event without an event field).

    test {
      let event = @sse.SseEvent::new("hello", id="1")
      assert_eq(event.event, "message")
      assert_eq(@sse.encode(event), b"id: 1\ndata: hello\n\n")
    }

    SseEvent::not_equal

    source
    fn SseEvent::not_equal(x : SseEvent, y : SseEvent) -> Bool

    Compares all event fields.

    SseEvent::to_repr

    source

    Debug representation of an event.

    SseParser

    source
    pub struct SseParser {
      // private fields
    }

    An incremental UTF-8 event-stream parser. Create one parser per stream.

    SseParser::feed

    source
    fn SseParser::feed(self : SseParser, bytes : BytesView) -> Array[SseEvent] raise 
    TransportError

    Feeds any byte chunk and returns complete events in order. Lines end at CR, LF or CRLF, also when a CRLF is split across chunks; a byte order mark is skipped only at the start of the stream, and malformed UTF-8 is replaced with U+FFFD. Feeding after finish is a no-op.

    An event over max_event_bytes raises TooLarge and ends the stream, which is refused rather than read without bound. When the same chunk completed events before it, those are returned first and the next feed or finish raises, so feed an empty chunk to observe it without waiting for more input.

    SseParser::finish

    source

    Ends the stream, discarding any incomplete event; always returns []. Repeated calls are harmless. Raises TooLarge if feed left an overflow unreported, so an oversized event at the end is never lost. last_event_id and reconnection_ms keep their values for the next connection.

    SseParser::last_event_id

    source
    fn SseParser::last_event_id(self : SseParser) -> String?

    The id a reconnecting client sends as Last-Event-ID: the last id field in effect at the most recent blank line, whether or not that blank line dispatched an event, as the WHATWG event-stream rules define the last event ID string. An id line of an event the stream never finished does not count. None before any blank line saw an id; Some("") after the stream reset it, in which case no header is sent.

    test {
      let parser = @sse.SseParser::new()
      assert_eq(parser.feed(b"id: 7\n\nid: 8\ndata: unfinished\n"), [])
      assert_eq(parser.finish(), [])
      assert_eq(parser.last_event_id(), Some("7"))
    }

    SseParser::new

    source
    fn SseParser::new(max_event_bytes? : Int) -> SseParser

    Creates an empty parser. max_event_bytes bounds one event: the raw bytes of its data lines, of the event and id lines in effect, and of the line being read, line terminators excluded. It defaults to 16 MiB. An event or id line that replaces an earlier one replaces its count too, and the id persists into later events, as the spec says, so it counts against each of them until replaced.

    test {
      let parser = @sse.SseParser::new()
      assert_eq(parser.feed(b"data: hello\n\n")[0].data, "hello")
      assert_eq(parser.finish(), [])
      let _ = @sse.SseParser::new(max_event_bytes=1_048_576)
    }

    SseParser::reconnection_ms

    source
    fn SseParser::reconnection_ms(self : SseParser) -> Int?

    The reconnection delay the stream asked for with its latest valid retry field, in milliseconds and capped at one hour, or None if it never sent one. As the WHATWG rules say, a retry field takes effect when it is read, even in an event that is never dispatched.

    test {
      let parser = @sse.SseParser::new()
      assert_eq(parser.feed(b"retry: 2500\n\n"), [])
      assert_eq(parser.reconnection_ms(), Some(2500))
    }

    SseReader

    source
    pub struct SseReader {
      // private fields
    }

    A pull reader over one event-stream body. Complete events from a chunk precede any overflow in that chunk; pending overflow is checked before reading again, even when the peer sends nothing more. Use one reader per source, without concurrent pulls, and close it when stopping early.

    SseReader::close

    source
    fn SseReader::close(self : SseReader) -> Unit

    Closes the source once and discards queued events, including before the first pull. Later pulls return EOF without touching the source.

    SseReader::each

    source
    async fn SseReader::each(self : SseReader, receive : async (SseEvent) -> Unit, stop? : (SseEvent) -> Bool) -> Unit

    Consumes events until EOF or a stop predicate matches, without delivering the matching sentinel. Closes the source on every exit, including callback failure and cancellation. An intentional stop discards queued events and any pending overflow, preserving protocol-sentinel behavior.

    SseReader::new

    source
    fn SseReader::new(source : &
    BodyStream    , max_event_bytes? : Int) -> SseReader

    Takes ownership of a source without reading it. Each event has the same byte bound as SseParser (16 MiB by default); queued events come from at most one source chunk. EOF, failure, cancellation and close release the source. each also closes it when its callback fails or a sentinel stops it.

    test {
      let d = @driver.Driver::new()
      let source = @sim.Feed::bytes(d, [b"data: hello\n\n"])
      let reader = @sse.SseReader::new(source)
      let events = []
      d.run(() => reader.each(event => events.push(event.data)))
      assert_eq(events, ["hello"])
      assert_eq(source.closes(), 1)
    }

    SseReader::next_event

    source

    Pulls the next complete event. EOF is permanent, incomplete final events are discarded, and transport failures propagate unchanged. Closes the source on EOF, failure or cancellation; a successful pull keeps it open.

    encode

    source
    fn encode(event : SseEvent) -> Bytes raise 
    TransportError

    Encodes one event as UTF-8, ending with a blank line. Data line endings become LF; each line gets its own data field. Event names and IDs cannot contain line breaks, IDs cannot contain NUL, and retries must be nonnegative. Invalid fields raise Invalid rather than injecting fields. An event of type message or "" gets no event field, since a stream gives that type to an event without one.

    Source Files