Server-Sent Events

The server/sse sub-package streams Server-Sent Events to clients. It gives you a Writer for emitting well-formed events on one connection and a Hub for broadcasting to many subscribers by topic. Import it as sse.

import "github.com/toaweme/http/server/sse"

Streaming a topic to a client

The quickest path is ServeStream. It subscribes the request to a topic, streams every event published to that topic, sends a heartbeat so proxies do not idle out the connection, and returns when the client disconnects.

hub := sse.NewHub()

r.Get("/stream", func(w http.ResponseWriter, req *http.Request) {
	_ = sse.ServeStream(w, req, hub, "updates") // blocks until the client leaves
})

Publish from anywhere that holds the hub, and every subscriber on that topic receives the event.

hub.Publish("updates", sse.Event{Type: "tick", Data: "hello"})

An Event carries an optional ID, a Type that maps to the SSE event: field, and a Data payload. When you publish a struct rather than a string, JSONEvent marshals it into the data field for you.

ev, err := sse.JSONEvent("item.created", Item{ID: "1"})
if err != nil {
	return err
}
hub.Publish("updates", ev)

How the hub handles slow subscribers

Each subscriber holds a buffered channel. When a subscriber's buffer is full at delivery time, the hub considers it slow, closes its channel, and drops it rather than blocking the producer. The subscriber sees a closed-channel receive and can decide to reconnect. Subscribers(topic) reports the current count, which is a cheap way to skip producing events nobody is listening for.

Writing events directly

When you drive a stream yourself rather than through a hub, NewWriter wraps a response writer. It returns an error when the writer cannot flush, since SSE is impossible without flushing. Start sends the streaming headers, Write emits one event, and Ping sends a heartbeat comment.

sw, err := sse.NewWriter(w)
if err != nil {
	http.Error(w, err.Error(), http.StatusInternalServerError)
	return
}
sw.Start()
for ev := range events {
	if err := sw.Write(ev); err != nil {
		return // client disconnected
	}
}

Subscribing without ServeStream

For full control over the receive loop, Subscribe returns a channel and a cancel func. Cancel through the returned func or by cancelling the context you pass, which unsubscribes and frees the channel.

ch, cancel := hub.Subscribe(r.Context(), "updates", 256)
defer cancel()
for ev := range ch {
	// forward ev however you need
}