Documentation
¶
Overview ¶
Package sse provides a broadcast hub for Server-Sent Events: fan-out to connected clients, a replay ring with monotonic event IDs so a reconnecting client resumes via the standard Last-Event-ID header, proxy-defensive response headers, keepalive comments, an optional concurrent-client cap, per-subscriber topic filtering, and a shutdown drain gate.
The Hub owns broadcast state; Serve adapts one HTTP request into a subscriber. Events carry pre-marshaled bytes (the hub does no JSON), an optional SSE event name (the `event:` field, for clients listening with addEventListener), and an optional topic that scopes delivery to subscribers that asked for it.
Delivery model ¶
Every published event is assigned a monotonically increasing ID, appended to a fixed-size replay ring, and fanned out to each subscriber's buffered channel. A subscriber that stops draining (a stalled client, a dead TCP peer behind a proxy) has its connection cancelled rather than the hub blocking: the standard EventSource client reconnects automatically, presents Last-Event-ID, and the ring replays what it missed. Replay and registration happen under one lock, so the replayed frames and the live channel are gap-free and overlap-free.
A reconnect that presents an ID older than the ring's floor means events were lost; the OnConnect hook receives the current (floor, head) bounds so an application can hand the client that information and let it refetch authoritative state.
The package has zero dependencies beyond the standard library.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Event ¶
type Event struct {
// Name is the SSE `event:` field. Empty emits an unnamed event, which an
// EventSource client receives via onmessage; a non-empty name requires
// addEventListener(name, ...) on the client.
Name string
// Topic scopes delivery: a topic-filtered subscriber receives an event
// only when the topics match. An event with an empty Topic is a broadcast
// delivered to every subscriber regardless of filter.
Topic string
// Data is the pre-marshaled payload for the `data:` field. The hub does
// no serialization of its own. A payload containing newlines is emitted
// as one `data:` line per line, per the SSE specification.
Data []byte
}
Event is one server-sent event to broadcast.
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub is a broadcast fan-out for Server-Sent Events with a replay ring. The zero value is not usable; construct with NewHub. A single Hub is safe for concurrent use by any number of publishers and subscribers.
func NewHub ¶
NewHub returns a Hub configured by the supplied options. A nil option is skipped, matching the root package's option convention.
func (*Hub) Bounds ¶
Bounds returns the replay window: floor is the oldest event ID still in the ring, head the newest ID assigned. Both are 0 before the first publish. A reconnecting client whose last-seen ID is below floor has missed events and should refetch authoritative state.
func (*Hub) Buffered ¶ added in v1.6.0
func (h *Hub) Buffered() []ReplayEvent
Buffered returns a snapshot of the replay ring, oldest first — the inspection surface for diagnostics (a debug endpoint, a test asserting what was published). Callers filter by ID or Topic themselves; the Last-Event-ID replay in Serve does not go through this method.
func (*Hub) ClientCount ¶
ClientCount returns the number of connected subscribers.
func (*Hub) Publish ¶
Publish assigns the event its ID, appends it to the replay ring, and fans it out to every matching subscriber. A subscriber whose buffer is full is evicted (its connection cancelled) rather than blocking the hub; the client's EventSource reconnects and resumes from Last-Event-ID. Publish on a nil Hub is a no-op, so an optional hub can be threaded without guards.
func (*Hub) Serve ¶
func (h *Hub) Serve(w http.ResponseWriter, r *http.Request, opts ...ServeOption)
Serve subscribes the request to the hub and streams events until the client disconnects, the request context ends, the subscriber is evicted, or the hub shuts down. It owns the response headers, Last-Event-ID replay, keepalive comments, and frame encoding.
It answers 503 while the hub is draining or over the client cap, and 500 when the ResponseWriter cannot stream: no http.Flusher reachable either directly or through an Unwrap() chain (the http.ResponseController discovery rule), so wrapping middleware that forwards flushes via Unwrap keeps streaming intact.
func (*Hub) SetMaxClients ¶ added in v1.7.0
SetMaxClients replaces the concurrent-subscriber cap at runtime (0 or negative = unlimited), for applications whose cap is hot-reloadable configuration: rebuilding the hub for a new cap would drop the replay ring and cancel every connected client. The cap is enforced atomically at admission (same critical section as subscribe), so a burst of concurrent connects cannot overshoot it. Lowering the cap does not evict already-connected clients; natural churn brings the count down. Safe for concurrent use.
type Option ¶
type Option func(*config)
Option configures a Hub at construction.
func WithClientBuffer ¶
WithClientBuffer sets the per-subscriber channel capacity (default: the replay ring size). A subscriber that falls this many events behind is evicted and relies on reconnect + replay.
func WithKeepalive ¶
WithKeepalive sets the interval between `: keepalive` comments (default 15s, below common proxy idle timeouts: nginx 60s, ALB 120s). A non-positive value disables keepalives.
func WithLogger ¶
WithLogger sets the slog.Logger for connect/disconnect/eviction diagnostics (default slog.Default()).
func WithMaxClients ¶
WithMaxClients caps concurrent subscribers; Serve answers 503 beyond it. 0 (the default) means unlimited.
func WithReplay ¶
WithReplay sets the replay ring capacity (default 256). The client-side delivery buffer is sized to match unless WithClientBuffer overrides it. A capacity of 0 disables replay: events still carry IDs, but a reconnect starts from live traffic only.
type ReplayEvent ¶
ReplayEvent is one buffered event with its assigned ID, as returned by Buffered.
type ServeOption ¶
type ServeOption func(*serveConfig)
ServeOption configures one Serve call.
func OnConnect ¶
func OnConnect(fn func(w *Writer, floor, head uint64) error) ServeOption
OnConnect installs a hook that runs after the subscriber is registered and after any Last-Event-ID replay, before live delivery begins. It receives the replay bounds so the application can write a handshake carrying them (letting the client detect a replay gap) and any initial per-client state. When no hook is installed, Serve writes a bare `: connected` comment.
func WithTopic ¶
func WithTopic(topic string) ServeOption
WithTopic subscribes this client to broadcasts plus events scoped to the given topic. An empty topic (the default) receives everything.
type Writer ¶
type Writer struct {
// contains filtered or unexported fields
}
Writer lets an OnConnect hook write frames onto the stream before live delivery starts. Writes are unflushed until the hook returns (Serve flushes once after the hook).
Source Files