Help us shape the future of Go! Take the 2026 Go Developer Survey

sse

package module
v1.2.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 23, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

README

sse

Go Reference npm JSR Mutation Mutation (TS)

sse keeps browser tabs in step with your Go server's Server-Sent Events through sleep, reconnects and restarts, with a Go hub and a matching TypeScript browser client.

A native EventSource cannot tell a silent stream from a dead one, so a tab that slept can silently stop updating. The sse hub and client replace the cursors, heartbeats and resync code you would write around that. The Go module needs Go 1.27.1 or later and one dependency, webhttp. The TypeScript package has no runtime dependencies. Both are Apache-2.0.

Why use it

sse is built for a Go backend that streams live state to browser tabs, where a missed event must never go unnoticed.

  • Every published frame carries an <epoch>:<offset> cursor. A reconnecting client gets every frame it missed, in order, or none and is told to reconcile.
  • DigestHandler tells a returning client which items changed or were removed, so it refetches only those, or everything after a server restart or a resolver failure.
  • The browser client checks liveness on received bytes and closes a hidden tab's stream until it returns.
  • One SharedWorker can carry one stream for all of a browser profile's tabs.
  • New returns an error for unsafe settings, such as a replay buffer with no age limit.

Consider Centrifugo if you want a separate real-time server that any backend publishes to. It speaks WebSocket, SSE and gRPC, recovers channel history on reconnect, and scales across nodes with Redis, PostgreSQL or Nats.

Install

go get github.com/cplieger/sse@latest
npx jsr add @cplieger/sse  # or: npm i @cplieger/sse

Usage

package main

import (
	"log"
	"net/http"
	"time"

	"github.com/cplieger/sse"
)

func main() {
	hub := sse.MustNew(
		sse.WithReplay(1024),
		sse.WithReplayTTL(10*time.Minute),
		sse.WithReplyMaxEvents(256),
	)

	mux := http.NewServeMux()
	mux.HandleFunc("GET /api/events", func(w http.ResponseWriter, r *http.Request) {
		hub.Serve(w, r, sse.WithTopic(r.URL.Query().Get("room")))
	})

	go func() {
		for t := range time.Tick(time.Second) {
			data := []byte(`{"at":"` + t.Format(time.RFC3339) + `"}`)
			if _, err := hub.Publish(sse.Event{Name: "tick", Data: data}); err != nil {
				log.Print(err)
			}
		}
	}()

	log.Fatal(http.ListenAndServe(":8080", mux))
}

This hub keeps the last 1024 frames for up to 10 minutes and sends at most 256 of them to one reconnecting client. MustNew panics on a refused setting. New takes the same options and returns an error that wraps ErrConfig instead. The browser side of the same endpoint is the createStream example in web/README.md.

Three compiled examples in example_test.go cover the next steps, and go test keeps them true:

  • ExampleHub_Serve writes initial state from an OnConnect hook, resumes a client from its cursor and shuts the hub down.
  • ExampleHub_DigestHandler mounts DigestHandler behind webhttp.RouteTimeout, which bounds your resolver.
  • ExampleNew shows a refused setting.

Publish refuses a frame over 1 MiB with ErrFrameTooLarge and invalid UTF-8 with ErrInvalidUTF8, before any state changes. For a large payload, publish a small frame that tells the client to fetch it. With webhttp, call Shutdown(ctx) from the WithPreDrain hook of webhttp.Run, so streams end before the HTTP server drains.

API

  • New, MustNew and twelve With* options build a hub. ErrConfig wraps every refusal.
  • Serve, with WithTopic, OnConnect and WithClientTag, streams one request. Writer.Event writes initial state from the hook.
  • Publish sends a frame and returns its offset, or ErrFrameTooLarge or ErrInvalidUTF8.
  • DigestHandler, a Resolver and two WithDigest* options answer a client that reconciles.
  • Position, Snapshot, ClientCount, QueuedFrames, SetMaxClients and Shutdown inspect and stop the hub.
  • Hello, Verdict, Cursor, ParseCursor, Wire, MaxOffset and MaxFrameBytes describe the wire. PresenceEvent is what the presence hook receives.
  • ssetest holds Serve, ReadFrames, FrameReader, Recorder and Fixture for your tests.
  • The TypeScript package exports createStream, createVersionMap, createDigestClient, createWorkerHost and attachToWorker, plus its parser and state machine.

The full reference is on pkg.go.dev and JSR. Running the hub covers every option and method.

A client resumes exactly or reconciles

Every connection opens with an sse:hello frame. Its resumed field is true only when the hub still holds every frame the client missed and their number is within the WithReplyMaxEvents cap. Those frames then follow the hello in order. Otherwise the hub replays nothing, and the client asks your server what changed. The hello's verdict says why, for logs and counters.

The hub runs inside one Go process and keeps its replay buffer in memory. Two processes never share a buffer. A client that falls too far behind gets an sse:reset frame and is dropped, so Publish never waits for it. The epoch is minted when the hub is created. A client that reconnects after a restart, or to another process, presents a cursor from another epoch and reconciles.

To reconcile, the client posts the versions it holds to DigestHandler. Your Resolver answers each item's current version, and the handler replies with what changed or was removed. Mount it and the Serve route behind your own authentication and cross-origin checks, because the hub performs neither.

The TypeScript client treats a stream without a valid hello as a failed connection. A native EventSource can read the stream and send Last-Event-ID. Your code must then read the sse:hello frame and reconcile when resumed is false, and it gets none of the client's liveness checks.

The wire contract lists the hello's fields, every verdict, the digest JSON and its refusals.

The browser client follows the tab

createStream owns the connection over fetch and ReadableStream, so it can send headers and see the response status. It presents the cursor it holds and treats the stream as dead after max(3 × keepalive, 15s) with no bytes, 45 seconds at the default keepalive. It closes a hidden tab's stream after 60 seconds and reopens it when the tab is shown. It reconnects with full-jitter backoff. While your revalidate callback runs, it holds incoming frames and then delivers them in order.

The client needs Chrome 98, Firefox 97 or Safari 15.4 or later. SharedWorker is optional, and each tab falls back to its own stream where it is missing. The DOM readers are injectable, so the same client runs in Node. web/README.md documents it in full.

Documentation

  • Running the hub covers every option, serving, publishing, presence and the test helpers, for anyone wiring the hub into a server.
  • The wire contract describes the cursor, the stream, the hello and the digest, for anyone writing a client or reading a stream by hand.
  • web/README.md documents the TypeScript client.

Credits

  • The cursor, the hello's verdict and the rule that only resumed: true resumes follow the recovery handshake of Centrifugo.
  • The browser client's parse loop follows eventsource-parser.
  • The silence watchdog's max(3 × keepalive, 15s) follows Yaffle/EventSource.
  • The client's version map follows IMAP CONDSTORE, RFC 7162. A new epoch discards every cached version, as a UIDVALIDITY change does.
  • The Go module's JSON error responses come from webhttp, a module by the same author that uses only the standard library.

Contributing

See CONTRIBUTING.md.

Disclaimer

This project is built with care and follows security best practices, but it is intended for personal / self-hosted use. No guarantees of fitness for production environments. Use at your own risk.

This project was built with AI-assisted tooling using Claude, GPT, and Kiro. The human maintainer defines architecture, supervises implementation, and makes all final decisions.

License

Apache-2.0. See LICENSE. Third-party attributions are in web/THIRD_PARTY_NOTICES.md.

Documentation

Overview

Package sse is a broadcast hub for Server-Sent Events whose clients resume from an exact position: every frame carries an "<epoch>:<offset>" id, every connection opens with a hello that says whether the presented cursor was honoured, and a hub refuses at construction any retention it could not keep.

A Hub owns the ring and the subscriber set; Serve adapts one HTTP request into a subscriber. The client presents its last id as Last-Event-ID and the hello answers with the epoch, the ring's floor and head, the keepalive and a Verdict: Resumed is true only when the ring covers every missed frame inside the reply cap, and a fresh connection, a cursor below the floor, another epoch's cursor or a malformed one all yield false, so the client reconciles.

Every timing constant both halves share is written once in timing.json and pinned by a test in each language; the TypeScript client is published as @cplieger/sse from web/ and documented in web/README.md.

Index

Examples

Constants

View Source 
const MaxFrameBytes = 1 << 20

MaxFrameBytes is the largest encoded frame, terminating blank line included. Publish and Writer.Event return ErrFrameTooLarge above it.

View Source 
const MaxOffset uint64 = 1<<53 - 1

MaxOffset is the largest offset ever emitted or accepted: the JavaScript exact-integer bound, so a peer holding offsets as Number never rounds one.

View Source 
const Wire = 1

Wire is the revision of the stream contract every hello declares.

Variables

View Source 
var (
	// ErrFrameTooLarge is returned by Publish and Writer.Event when the encoded
	// frame would exceed MaxFrameBytes; it is wrapped with the size and the cap.
	ErrFrameTooLarge = errors.New("sse: frame exceeds MaxFrameBytes")
	// ErrInvalidUTF8 is returned by Publish and Writer.Event when Data is not
	// valid UTF-8; it is wrapped with the byte offset of the first invalid byte.
	ErrInvalidUTF8 = errors.New("sse: data is not valid UTF-8")
)
View Source 
var ErrConfig = errors.New("sse: incoherent configuration")

ErrConfig is wrapped by every construction-time refusal New returns; the message names the offending option or pair.

View Source 
var ErrCursor = errors.New("sse: invalid cursor")

ErrCursor is returned by ParseCursor, wrapped with the reason, for any non-empty string that is not a well-formed "<epoch>:<offset>".

Functions

This section is empty.

Types

type Cursor 

type Cursor struct {
	Epoch  string
	Offset uint64
}

Cursor is a stream position: the hub epoch that issued Offset, and Offset itself. The zero Cursor means no position.

func ParseCursor 

func ParseCursor(s string) (Cursor, error)

ParseCursor parses a Last-Event-ID value. The empty string is no cursor and yields the zero Cursor with a nil error. Anything else must be exactly 16 lowercase hex characters, a colon and a decimal offset with no leading zero at most MaxOffset; a bare integer is invalid.

func (Cursor) String 

func (c Cursor) String() string

String renders the cursor as "<epoch>:<offset>", the id: field's value.

type DigestOption 

type DigestOption func(*digestConfig)

DigestOption configures one DigestHandler.

func WithDigestMaxBody 

func WithDigestMaxBody(n int64) DigestOption

WithDigestMaxBody caps the request body in bytes (default 512 KiB, sized so a full default batch of worst-case escaped refs and versions cannot 413). Non-positive keeps the default.

func WithDigestMaxSubjects 

func WithDigestMaxSubjects(n int) DigestOption

WithDigestMaxSubjects caps the subjects one request may carry (default 256); a larger request is 400. Non-positive keeps the default.

type Event 

type Event struct {
	// Topic scopes delivery: "" reaches every client, any other value only the
	// clients subscribed to that topic.
	Topic string
	// Name is the event: field. It may be empty (no event: line; a browser
	// dispatches the frame as "message"), must not start with "sse:", and must
	// not contain CR or LF.
	Name string
	// Data is the data: payload, UTF-8 text split on CRLF, CR and LF into one
	// data: line each. Publish takes ownership of the slice: the ring retains
	// it until eviction and replay reads it, so a caller that writes to it
	// afterwards changes replay bytes after the UTF-8 and size checks passed.
	// Marshal fresh per publish.
	Data []byte
}

Event is one server-sent event to publish.

type Held 

type Held struct {
	Subject
	Version string
}

Held is a Subject with the version the client presents for it.

type Hello 

type Hello struct {
	Epoch          string  `json:"epoch"`
	Verdict        Verdict `json:"verdict"`
	KeepaliveEvent string  `json:"keepalive_event"`
	Floor          uint64  `json:"floor,string"`
	Head           uint64  `json:"head,string"`
	Wire           int     `json:"wire"`
	KeepaliveMS    int     `json:"keepalive_ms"`
	Resumed        bool    `json:"resumed"`
}

Hello is the handshake frame's payload, written once per connection right after the retry: field and handed to the OnConnect hook unchanged. Floor and Head are decimal strings on the wire because the peer is a JavaScript client.

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 New. Safe for concurrent use by any number of publishers and subscribers.

func MustNew 

func MustNew(opts ...Option) *Hub

MustNew is New for package-level construction in main; it panics with the wrapped ErrConfig message on an incoherent option set.

func New 

func New(opts ...Option) (*Hub, error)

New returns a Hub configured by opts, or an error wrapping ErrConfig for an incoherent option set. A nil option is skipped.

Example 
package main

import (
	"errors"
	"fmt"
	"time"

	"github.com/cplieger/sse"
)

func main() {
	hub, err := sse.New(
		sse.WithReplay(1024),
		sse.WithReplayTTL(10*time.Minute),
		sse.WithReplyMaxEvents(256),
	)
	if err != nil {
		panic(err)
	}

	offset, err := hub.Publish(sse.Event{Name: "notify", Data: []byte(`{"n":1}`)})
	fmt.Println(offset, err)

	pos := hub.Position()
	fmt.Println(pos.Floor, pos.Head)

	// A ring without an age bound is refused at construction.
	_, err = sse.New(sse.WithReplay(1024))
	fmt.Println(errors.Is(err, sse.ErrConfig))
}

Output:
1 <nil>
1 1
true

func (*Hub) ClientCount 

func (h *Hub) ClientCount() int

ClientCount returns the number of subscribed clients: registered and not yet reset as slow. A connection inside its OnConnect hook is counted; one that Publish evicted is not, even while its goroutine is still writing the reset. After Shutdown it counts the goroutines still draining.

func (*Hub) DigestHandler 

func (h *Hub) DigestHandler(resolve Resolver, opts ...DigestOption) http.Handler

DigestHandler answers POST application/json digest requests: which of the client's held subjects changed or were removed, pinned to the hub's epoch and head. Mount it inside the application's authentication and cross-origin middleware; it performs neither check itself. The response is must_refetch when the request epoch is absent or not this hub's, when resolve fails, or when its output does not match the request by key; every well-formed request is answered 200.

Example 
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http/httptest"
	"strings"
	"time"

	"github.com/cplieger/sse"
	"github.com/cplieger/webhttp/v3"
)

func main() {
	hub := sse.MustNew()
	current := map[sse.Subject]string{{Kind: "chat", Ref: "c1"}: "7"}
	resolve := func(_ context.Context, held []sse.Held) ([]sse.State, error) {
		states := make([]sse.State, 0, len(held))
		for _, h := range held {
			version, ok := current[h.Subject]
			if !ok {
				states = append(states, sse.State{Subject: h.Subject, Status: sse.StatusGone})
				continue
			}
			states = append(states, sse.State{Subject: h.Subject, Version: version})
		}
		return states, nil
	}
	srv := httptest.NewServer(webhttp.RouteTimeout(hub.DigestHandler(resolve), 10*time.Second, "digest timed out"))
	defer srv.Close()

	body := fmt.Sprintf(`{"epoch":%q,"subjects":[{"kind":"chat","ref":"c1","version":"5"},{"kind":"chat","ref":"c2","version":"1"}]}`, hub.Position().Epoch)
	resp, err := srv.Client().Post(srv.URL, "application/json", strings.NewReader(body))
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	var out struct {
		Changed []sse.Held `json:"changed"`
		Removed []struct {
			sse.Subject
			Reason sse.Status `json:"reason"`
		} `json:"removed"`
		MustRefetch bool `json:"must_refetch"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		panic(err)
	}
	fmt.Println(resp.StatusCode, out.MustRefetch)
	fmt.Println(out.Changed)
	fmt.Println(out.Removed)
}

Output:
200 false
[{{chat c1} 7}]
[{{chat c2} gone}]

func (*Hub) Position 

func (h *Hub) Position() Position

Position returns the replay window after evicting expired entries, so its Floor is the one a subscription at this instant would be judged against.

func (*Hub) Publish 

func (h *Hub) Publish(ev Event) (uint64, error)

Publish assigns the event the next offset, appends it to the ring and fans it out to every matching client without blocking; a client whose buffer is full is reset as slow and dropped. It returns the offset, or (0, nil) on a nil *Hub and for a valid frame after Shutdown (dropped). A Name outside Event's grammar panics; Data that is not UTF-8 returns ErrInvalidUTF8 and an encoded frame above MaxFrameBytes returns ErrFrameTooLarge, and neither refusal consumes an offset or touches the ring, before or after Shutdown.

func (*Hub) QueuedFrames 

func (h *Hub) QueuedFrames() int

QueuedFrames returns the sum of frames queued across client channels.

func (*Hub) Serve 

func (h *Hub) Serve(w http.ResponseWriter, r *http.Request, opts ...ServeOption)

Serve subscribes the request to the hub and streams until the peer leaves, the request context ends, the client is reset as slow, a write fails, or the hub shuts down. It owns the response headers, the retry: field, the hello, the Last-Event-ID replay, keepalives and the sse:reset frame.

It answers 503 sse_unavailable over the client cap or after Shutdown, and 500 streaming_unsupported when no http.Flusher is reachable through the ResponseWriter or its Unwrap chain.

Example 
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"time"

	"github.com/cplieger/sse"
	"github.com/cplieger/sse/ssetest"
)

func main() {
	hub := sse.MustNew(sse.WithReplay(64), sse.WithReplayTTL(10*time.Minute))

	mux := http.NewServeMux()
	mux.HandleFunc("GET /api/events", func(w http.ResponseWriter, r *http.Request) {
		hub.Serve(w, r,
			sse.OnConnect(func(w *sse.Writer, h sse.Hello) error {
				return w.Event("connected", fmt.Appendf(nil, `{"resumed":%t}`, h.Resumed))
			}),
			sse.WithClientTag(r.Header.Get("SSE-Client")),
		)
	})
	srv := httptest.NewServer(mux)
	defer srv.Close()

	// A first connection presents no cursor: the hello is fresh.
	first, err := open(srv, "")
	if err != nil {
		panic(err)
	}
	firstStream := ssetest.NewFrameReader(first.Body)
	frames, err := firstStream.Read(3)
	if err != nil {
		panic(err)
	}
	fmt.Println(frames[0].Event, frames[0].Data)
	fmt.Println(frames[1].Event, describe(frames[1].Data))
	fmt.Println(frames[2].Event, frames[2].Data)

	if _, err := hub.Publish(sse.Event{Name: "notify", Data: []byte(`{"n":1}`)}); err != nil {
		panic(err)
	}
	live, err := firstStream.Read(1)
	if err != nil {
		panic(err)
	}
	cursor, err := sse.ParseCursor(live[0].ID)
	if err != nil {
		panic(err)
	}
	fmt.Println(live[0].Event, cursor.Offset, live[0].Data)
	first.Body.Close()

	// A frame published while the client is away is replayed on resume.
	if _, err := hub.Publish(sse.Event{Name: "notify", Data: []byte(`{"n":2}`)}); err != nil {
		panic(err)
	}
	second, err := open(srv, cursor.String())
	if err != nil {
		panic(err)
	}
	defer second.Body.Close()
	secondStream := ssetest.NewFrameReader(second.Body)
	frames, err = secondStream.Read(4)
	if err != nil {
		panic(err)
	}
	fmt.Println(frames[1].Event, describe(frames[1].Data))
	fmt.Println(frames[2].Event, frames[2].ID == sse.Cursor{Epoch: cursor.Epoch, Offset: 2}.String(), frames[2].Data)
	fmt.Println(frames[3].Event, frames[3].Data)

	// Shutdown ends every stream with a reset frame naming the reason.
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	fmt.Println(hub.Shutdown(ctx))
	reset, err := secondStream.Read(1)
	if err != nil {
		panic(err)
	}
	fmt.Println(reset[0].Event, reset[0].Data)
}

func open(srv *httptest.Server, cursor string) (*http.Response, error) {
	req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, srv.URL+"/api/events", nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("Accept", "text/event-stream")
	req.Header.Set("SSE-Client", "tab-1")
	if cursor != "" {
		req.Header.Set("Last-Event-ID", cursor)
	}
	return srv.Client().Do(req)
}

func describe(hello string) string {
	var h sse.Hello
	if err := json.Unmarshal([]byte(hello), &h); err != nil {
		panic(err)
	}
	return fmt.Sprintf("%s resumed=%t head=%d", h.Verdict, h.Resumed, h.Head)
}

Output:
retry 1500
sse:hello fresh resumed=false head=0
connected {"resumed":false}
notify 1 {"n":1}
sse:hello resumed resumed=true head=2
notify true {"n":2}
connected {"resumed":true}
<nil>
sse:reset {"reason":"shutdown"}

func (*Hub) SetMaxClients 

func (h *Hub) SetMaxClients(n int)

SetMaxClients replaces the client cap at runtime (0 or negative means unlimited). Lowering it evicts nobody; admission enforces it.

func (*Hub) Shutdown 

func (h *Hub) Shutdown(ctx context.Context) error

Shutdown refuses new subscriptions, signals every client to write its sse:reset and waits for the stream goroutines to return, bounded by ctx. Per client the wait is the remainder of a running OnConnect hook plus max(write timeout, 250ms). Returns ctx.Err() when ctx expires first.

func (*Hub) Snapshot 

func (h *Hub) Snapshot() []ReplayEvent

Snapshot returns a copy of the ring, oldest first, with each entry's Data cloned so the caller cannot change replay bytes. A diagnostic surface.

type Option 

type Option func(*config) error

Option configures a Hub at construction. An option refuses its own value on the spot; rules that compare two options run after every option has been applied, so option order never changes the verdict.

func WithClientBuffer 

func WithClientBuffer(n int) Option

WithClientBuffer sets the per-client channel capacity (default max(ring capacity, 256)). A client that falls this many frames behind is reset as slow. Refused at or below zero.

func WithKeepalive 

func WithKeepalive(d time.Duration) Option

WithKeepalive sets the keepalive interval (default 15s), announced to the client as hello.keepalive_ms. Refused below one millisecond.

func WithKeepaliveEvent 

func WithKeepaliveEvent(name string) Option

WithKeepaliveEvent names the keepalive frame's event (default "sse:keepalive"). The empty string selects the comment form ": keepalive", invisible to EventSource consumers. A name holding CR or LF is refused.

func WithLogger 

func WithLogger(l *slog.Logger) Option

WithLogger sets the logger for connection and eviction diagnostics (default slog.Default()). A nil logger keeps the default.

func WithMaxClients 

func WithMaxClients(n int) Option

WithMaxClients caps concurrent clients; Serve answers 503 beyond it. Zero or negative means unlimited.

func WithPresence 

func WithPresence(fn func(PresenceEvent)) Option

WithPresence installs the one hook that receives every PresenceEvent (nil removes it). It runs on the arriving or departing stream's own goroutine, after the hello flush or after the client is unregistered, never under the hub lock, so a slow hook delays only that stream. A panic in it is recovered and logged at Error.

func WithReconnectDelay 

func WithReconnectDelay(d time.Duration) Option

WithReconnectDelay sets the stream's retry: field (default 1500ms), the wait a client observes before reconnecting. Refused below one millisecond, since 0 on the wire means reconnect at once.

func WithReplay 

func WithReplay(n int) Option

WithReplay sets the ring capacity in events (default 0, no replay). A non-zero capacity requires WithReplayTTL. Negative is refused.

func WithReplayMaxBytes 

func WithReplayMaxBytes(n int) Option

WithReplayMaxBytes caps the sum of encoded frame bytes the ring holds (default 64 × MaxFrameBytes); the oldest entries are evicted when an append would exceed it. Refused below MaxFrameBytes.

func WithReplayTTL 

func WithReplayTTL(d time.Duration) Option

WithReplayTTL sets the maximum age of a ring entry (no default; required when WithReplay is non-zero). Refused at or below zero, and refused below max(3 × keepalive, 15s) + 30s, the window a client needs to reconnect after a half-open stream.

func WithReplyMaxEvents 

func WithReplyMaxEvents(n int) Option

WithReplyMaxEvents caps the replay one connection may receive (default min(ring capacity, 256)); a resume needing more is gap_budget. Refused when negative or above the ring capacity.

func WithWriteTimeout 

func WithWriteTimeout(d time.Duration) Option

WithWriteTimeout bounds every write to a client (default 2 × keepalive). Refused at or below the keepalive interval.

type Position 

type Position struct {
	Epoch string
	Floor uint64
	Head  uint64
}

Position is the hub's replay window at one instant: the epoch, the oldest offset the ring can replay (0 when empty) and the newest offset published.

type PresenceCause 

type PresenceCause string

PresenceCause says why a client left, on a disconnected PresenceEvent. 

const (
	PresenceClosed     PresenceCause = "closed"
	PresenceDead       PresenceCause = "dead"
	PresenceEvicted    PresenceCause = "evicted"
	PresenceShutdown   PresenceCause = "shutdown"
	PresenceHookFailed PresenceCause = "hook_failed"
)

The causes a disconnected event carries. Closed and dead are the socket's account; evicted and shutdown are the two sse:reset reasons; hook_failed is the application ending its own connection from OnConnect.

type PresenceEvent 

type PresenceEvent struct {
	At    time.Time
	Kind  PresenceKind
	Topic string
	// Verdict is the hello's verdict on connected; empty on disconnected.
	Verdict Verdict
	// Cause is set on disconnected; empty on connected.
	Cause PresenceCause
	// Write names the failed write: "keepalive", "frame" or "hook" on a dead
	// event, "reset" on an evicted or shutdown event whose reset write failed.
	Write string
	Epoch string
	// Tag is the WithClientTag value, or "" when none was passed or it failed
	// the grammar. A presence table folds clients by it.
	Tag string
	// ClientID is process-unique and minted at subscribe, so two connects from
	// one browser are two clients.
	ClientID uint64
}

PresenceEvent is one arrival or departure as WithPresence reports it. A connected event is delivered once per Serve that flushed its hello, and exactly one disconnected event follows it; a Serve that failed before the hello produces neither. Dead means a write or flush failed with the socket still open, which the kernel may take minutes to notice: it is observability of socket death, not a liveness bound.

type PresenceKind 

type PresenceKind string

PresenceKind says whether a PresenceEvent is an arrival or a departure. 

const (
	PresenceConnected    PresenceKind = "connected"
	PresenceDisconnected PresenceKind = "disconnected"
)

The two kinds a PresenceEvent carries.

type ReplayEvent 

type ReplayEvent struct {
	At     time.Time
	Event  Event
	Offset uint64
}

ReplayEvent is one ring entry as Snapshot returns it.

type Resolver 

type Resolver func(ctx context.Context, held []Held) ([]State, error)

Resolver answers the current State of every held subject, exactly one per requested (Kind, Ref) and none the request did not carry. It runs on the request goroutine with the request's context and no library-level concurrency bound; the application bounds it in time and in flight. The principal comes from ctx however the application's middleware stored it.

type ServeOption 

type ServeOption func(*serveConfig)

ServeOption configures one Serve call.

func OnConnect 

func OnConnect(fn func(w *Writer, h Hello) error) ServeOption

OnConnect installs a hook that runs after the retry: field, the hello and the replay have been flushed, before live delivery. It receives the Hello the client received and writes initial-state frames through the Writer; an error ends the connection. The hook runs on the stream goroutine, holds no hub lock, and may call Publish and Position. Keep it short: a hook that runs past the client's hello timeout loses the tab to a reconnect.

func WithClientTag 

func WithClientTag(tag string) ServeOption

WithClientTag attaches an application-chosen presence key to the connection. The empty string means no tag, so a handler may pass the SSE-Client header through unconditionally; any other tag outside [A-Za-z0-9_-]{1,64} is treated as absent with one Warn naming its length and the class of failure, never the bytes.

func WithTopic 

func WithTopic(topic string) ServeOption

WithTopic subscribes the connection to broadcasts plus events scoped to topic. The empty topic (the default) receives everything.

type State 

type State struct {
	Subject
	Version string
	Status  Status
}

State is a resolver's answer for one requested subject.

type Status 

type Status string

Status is a resolver's answer for a subject it no longer serves. 

const (
	StatusCurrent   Status = ""
	StatusGone      Status = "gone"
	StatusForbidden Status = "forbidden"
)

StatusCurrent means the subject exists and Version is its current version; StatusGone means it no longer exists; StatusForbidden means it exists and the principal may not read it. In a multi-principal deployment the resolver decides whether the principal may learn the distinction and answers gone for both when it may not.

type Subject 

type Subject struct {
	Kind string
	Ref  string
}

Subject is one application-defined unit of state the digest compares: a Kind the application chooses and an opaque Ref within it.

type Verdict 

type Verdict string

Verdict is the hello's diagnostic account of how a presented cursor was judged. Clients branch on Hello.Resumed only; the verdict feeds logs and counters. 

const (
	VerdictFresh         Verdict = "fresh"
	VerdictResumed       Verdict = "resumed"
	VerdictGapFloor      Verdict = "gap_floor"
	VerdictGapBudget     Verdict = "gap_budget"
	VerdictGapAhead      Verdict = "gap_ahead"
	VerdictEpochChanged  Verdict = "epoch_changed"
	VerdictCursorInvalid Verdict = "cursor_invalid"
)

The seven verdicts. The three gap variants are kept distinct so a counter can tell a retention problem from a reply-cap problem from a client defect.

type Writer 

type Writer struct {
	// contains filtered or unexported fields
}

Writer is handed to OnConnect only; frames it writes carry no id. Serve always supplies it, so there is no constructor and no useful zero value.

func (*Writer) Event 

func (sw *Writer) Event(name string, data []byte) error

Event writes one frame as its own bounded write and flush; name may be empty. It returns ErrFrameTooLarge or ErrInvalidUTF8 exactly as Publish does (the size check has no id line here), panics on the same names, and returns the deadline error when the peer stopped reading, which the hook propagates.

Source Files

View all Source files

Directories

Path Synopsis
Package ssetest is the test seam between the sse hub and its clients: an httptest server for a hub, a frame reader for the wire, a recorder that answers http.ErrNotSupported to the deadline setters, and Fixture, the controllable server the TypeScript integration suites drive through the ssetest/cmd binary.
 Package ssetest is the test seam between the sse hub and its clients: an httptest server for a hub, a frame reader for the wire, a recorder that answers http.ErrNotSupported to the deadline setters, and Fixture, the controllable server the TypeScript integration suites drive through the ssetest/cmd binary.
cmd command
Command ssetest serves an ssetest.Fixture for the TypeScript integration suites.
 Command ssetest serves an ssetest.Fixture for the TypeScript integration suites.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.