Server-Sent Events

Router.SSE registers a route whose response is a stream rather than a body. It is the other half of what a WebSocket is usually reached for, and the simpler half: the server sends, the client listens, and a browser reads it natively with EventSource, reconnecting on its own when a stream drops.

// An event stream route is declared like any other and answers 200 rather than
// upgrading: the guards run and the input binds before a byte of the stream is
// written, and the handler owns the stream until it returns.
r.SSE("/items/stream", handlers.StreamItems,
    muzak.Summary("Follow every change to the items"),
    muzak.WithSSE(muzak.SSEOptions{
        // A stream that says nothing for long enough is closed by proxies that
        // believe it to be idle, so a comment goes out instead.
        KeepAlive: 15 * time.Second,
        // A browser reconnects on its own when a stream ends, and this is how
        // soon.
        Retry: 2 * time.Second,
    }))

func StreamItems(ctx *muzak.Context, _ muzak.Empty, stream *muzak.SSEStream[schemas.ItemOut]) error {
    store := muzak.From[*core.ItemStore](ctx)

    updates := store.Watch(stream.Context())

    for {
        select {
        case <-stream.Context().Done():
            // The client went away or the server is shutting down. Either way
            // the conversation is over, and it is not a failure.
            return nil
        case change := <-updates:
            item := schemas.ItemOut{ID: change.Item.ID, Name: change.Item.Name}
            if err := stream.Send(item); err != nil {
                return err
            }
        }
    }
}

The type parameter is the contract: nothing but an ItemOut can be sent, and the generated document describes the stream with that type, in the same way a handler's return type describes an ordinary response.

curl -N 'http://localhost:8080/items/stream'

retry: 2000

event: item_update
id: 1
data: {"id":"plumbus","name":"Plumbus"}

event: item_update
id: 2
data: {"id":"fleeb","name":"Fleeb"}

The retry field is sent once, before anything else, because SSEOptions.Retry was set on the route above: a client that loses the stream on the very first event already knows how long to wait. The response carries Content-Type: text/event-stream; charset=utf-8, Cache-Control: no-cache, no-transform and X-Accel-Buffering: no.

An ordinary route

Middleware runs, guards run, dependencies resolve, and the input is bound and validated before a byte of the stream is written. A request that fails any of that is answered with the usual JSON error and never becomes a stream at all.

The response header is written before the handler is called, which is what lets a client see the stream open immediately. Anything that decides whether to serve a stream at all therefore belongs in a guard or a dependency, where there is still a response to say it in.

The stream

Nothing on SSEStream takes a context, unlike WSConn, because a stream belongs to one request.

stream.Context() is derived from the request's own and is cancelled when the client disconnects, when the request ends, or when the server begins shutting down. That makes it the one thing a handler has to watch, and every send after it reports muzak.ErrSSEStreamEnded, so a handler that only sends ends on its own.

for update := range updates {
    if err := stream.Send(update); err != nil {
        return err
    }
}

A handler that returns ErrSSEStreamEnded is treated as a stream that finished rather than one that failed, so a client going away is not logged as an error. Test for it with errors.Is; the reason the stream ended is wrapped inside.

Writes are serialized, so any number of goroutines may write to one stream and each event goes out whole.

A stream is single use. The first failure ends it, and every later send reports that same error rather than writing into a response nobody is reading. Sends that report a mistake instead, such as an event name carrying a line break, leave the stream usable.

Events with more than data

type SSEEvent[Out any] struct {
    Name    string        // the type a browser dispatches it under
    ID      string        // what a reconnect resumes from
    Retry   time.Duration // how long to wait before reconnecting
    Comment string        // text no client acts on
    Data    *Out          // the value, encoded as JSON
    Text    string        // a payload written as it stands
}

// sendChange writes one change as an event a browser can listen for by name
// and resume from by identifier.
func sendChange(stream *muzak.SSEStream[schemas.ItemOut], change core.Change) error {
    item := schemas.ItemOut{ID: change.Item.ID, Name: change.Item.Name}
    return stream.SendEvent(muzak.SSEEvent[schemas.ItemOut]{
        Name: "item_update",
        ID:   strconv.Itoa(change.Seq),
        Data: &item,
    })
}

const source = new EventSource("/items/stream")
source.addEventListener("item_update", (event) => {
    const item = JSON.parse(event.data)
    // ...
})

Data is a pointer so that an event carrying nothing can be told from one carrying a zero value, and it cannot be combined with Text. An event setting both is refused.

Name and ID may not contain a line break. An event stream is a sequence of lines, so a break in one of those fields would end it and let whatever followed be read as fields of its own, which on a stream carrying one client's input to another is event forgery. Such a send is refused rather than repaired, and the stream stays usable.

Payloads that are not JSON

Text writes a payload as it stands, for a log line or for the sentinel some completion APIs end with. A stream whose events are all text declares muzak.Empty as its model, and the generated document then says there is no schema rather than describing one that does not exist.

// StreamChat answers a prompt one token at a time, which is the shape every
// chat completion API streams in.
func StreamChat(ctx *muzak.Context, in schemas.ChatIn, stream *muzak.SSEStream[muzak.Empty]) error {
    for word := range strings.SplitSeq(in.Text, " ") {
        select {
        case <-stream.Context().Done():
            // The client closed the tab, or the server is shutting down. There
            // is no one left to answer.
            return nil
        case <-time.After(tokenDelay):
        }
        if err := stream.SendEvent(muzak.SSEEvent[muzak.Empty]{Name: "token", Text: word}); err != nil {
            return err
        }
    }
    // The sentinel some clients expect at the end of a completion. It is text
    // rather than a value, which is what Text is for.
    return stream.SendEvent(muzak.SSEEvent[muzak.Empty]{Name: "done", Text: "[DONE]"})
}

A payload spanning several lines is written as several data lines and arrives whole, which is both what the format asks for and what stops a value from ending its own field.

A stream reached by POST

A stream is not tied to GET. Router.SSEHandle registers one for any method, which is what a protocol that streams its answer to a posted document needs, and there the input binds a request body like any other route.

r.SSEHandle(http.MethodPost, "/chat/stream", handlers.StreamChat,
    muzak.Summary("Answer a prompt one token at a time"),
    muzak.WithSSE(muzak.SSEOptions{
        // A completion is never quiet for long, so the keepalive is only there
        // for the pause before the first token.
        KeepAlive: 10 * time.Second,
    }))

// ChatIn is the prompt a chat stream answers.
//
// It is an ordinary JSON body: an event stream reached by POST binds and
// validates one exactly as any other route does.
type ChatIn struct {
    Text string `json:"text" doc:"The prompt to answer, one token at a time"`
}

// Validate bounds the prompt, and is applied before a byte of the stream is
// written.
func (in *ChatIn) Validate(v *muzak.Validation) {
    v.String(&in.Text).Trim().Required().MinLen(1).MaxLen(280)
}

curl -N -X POST 'http://localhost:8080/chat/stream' \
     -H 'Content-Type: application/json' -d '{"text":"what is a plumbus"}'

Resuming a dropped stream

A browser remembers the last identifier it saw and sends it back in Last-Event-ID when it reconnects. stream.LastEventID() reads it, and that is what turns a dropped connection into a stream that picks up where it left off rather than one that starts again.

func StreamItems(ctx *muzak.Context, _ muzak.Empty, stream *muzak.SSEStream[schemas.ItemOut]) error {
    store := muzak.From[*core.ItemStore](ctx)

    // A browser sends back the identifier of the last event it saw when its
    // EventSource reconnects. The value is the client's, so one that is not a
    // number is treated as no value at all rather than as an error.
    seen := 0
    if last := stream.LastEventID(); last != "" {
        if parsed, err := strconv.Atoi(last); err == nil {
            seen = parsed
        }
    }

    // Subscribing before the backlog is read is what stops a change made in
    // between from falling through the gap between the two.
    updates := store.Watch(stream.Context())
    for _, change := range store.Since(seen) {
        if err := sendChange(stream, change); err != nil {
            return err
        }
        seen = change.Seq
    }

    for {
        select {
        case <-stream.Context().Done():
            return nil
        case change := <-updates:
            if change.Seq <= seen {
                // Already sent from the backlog above.
                continue
            }
            if err := sendChange(stream, change); err != nil {
                return err
            }
            seen = change.Seq
        }
    }
}

LastEventID is a value the client controls, so treat it as input rather than as a cursor to be trusted.

What a stream bounds

A stream costs a connection and a goroutine for as long as a client cares to hold it.

Clearing the listener's read and write deadlines matters more than it sounds. A stream is a response that does not end, so it would otherwise die at ServerOptions.WriteTimeout however healthy it was, and the read deadline would cancel the request, and with it the stream, at ServerOptions.ReadTimeout and blame the client.

Options

Set them application-wide through AppOptions.SSE and narrow them for a router or a route with WithSSE. Layering works field by field, so a route that only lengthens the keepalive keeps the application's write timeout.

MaxStreams and MaxStreamsPerIP may only be set on the application: the resource they protect is the process, not a route, so a router or a route that sets either is refused when the application is built.

The keepalive exists because a proxy that sees an idle connection for long enough closes it, and because a silent stream is indistinguishable from a dead one.

What does not apply

Compression leaves an event stream alone. Holding events in a compressor's window until something forces them out is the one thing a stream cannot survive.

The origin check a WebSocket needs has no counterpart here. An EventSource is subject to the same-origin policy and to CORS like any other request, so AppOptions.CORS already governs it. See CORS.

Nothing a handler fails with is disclosed. The response header went out before the handler ran, so a failure ends the stream and the reason goes to the log.

Shutdown

Open streams are tracked, so a graceful shutdown ends every one of them and waits for the handlers, instead of waiting out its whole deadline once per stream.

Documentation

An SSE route is documented with a 200 response carrying text/event-stream, whose schema describes the data field of a single event. OpenAPI 3.1 has no way to say "many of these, one after another", and one event's data is the only part a client has to be able to decode. A stream whose model is muzak.Empty is documented with no schema at all.

Testing

func TestStreamItems(t *testing.T) {
    client := testclient.New(t, buildApp())

    stream := client.SSE("/items/stream")

    client.Post("/items/", testclient.JSON(map[string]string{"id": "plumbus", "name": "Plumbus"})).
        AssertStatus(http.StatusCreated)

    item := stream.Decode[schemas.ItemOut]()
    if item.ID != "plumbus" {
        t.Errorf("id = %q, want plumbus", item.ID)
    }
}

muzak.SSEDial is the reading half of the same engine, so a route is tested over a real connection rather than against a second implementation:

reader, _, err := muzak.SSEDial(ctx, "http://"+app.Addr()+"/items/stream", muzak.SSEDialOptions{})
if err != nil {
    return err
}
defer reader.Close()

for {
    message, err := reader.Next(ctx)
    if errors.Is(err, muzak.ErrSSEStreamEnded) {
        return nil
    }
    if err != nil {
        return err
    }
    item, err := message.Decode[schemas.ItemOut]()
    if err != nil {
        return err
    }
    _ = item
}

A response that is not 200 with a text/event-stream body is refused rather than parsed, because a stream reader that quietly accepts an HTML error page reports "no events" for what is actually a failure. SSEDialOptions carries the method, body, headers, a LastEventID to resume from, a read limit, a read timeout and KeepComments for seeing the keepalives.

Where to go next

WebSockets covers the two-way half, and Testing covers asserting on a stream.