Documentation
¶
Overview ¶
Package events owns the in-process Server-Sent Events hub for the mock's GET /events endpoint. The hub fans out events posted via POST /admin0/events to every connected subscriber whose event_type filter matches, backed by a bounded ring buffer that lets dropped subscribers resume via Last-Event-ID / ?from / ?from_timestamp.
The package wraps github.com/tmaxmax/go-sse: sse.Server is the HTTP handler, sse.Joe is the in-memory pub/sub provider, and recordingReplayer (this package) adds the (id, timestamp) index needed to translate ?from_timestamp into an event ID for the library's Last-Event-ID-driven Replay path.
Index ¶
Constants ¶
const DefaultKeepAliveInterval = 15 * time.Second
DefaultKeepAliveInterval is the cadence at which a `:keep-alive` comment is broadcast to every connected subscriber. 15s matches what most SSE deployments use; the library doesn't auto-emit.
const DefaultReconnectHint = 3 * time.Second
DefaultReconnectHint is the value sent in the SSE `retry:` field on connect, telling clients how long to wait before reconnecting after a disconnect (Auth0's Events API sends one too). A value <= 0 omits the hint, so clients fall back to their built-in default.
Variables ¶
This section is empty.
Functions ¶
func SetKeepAliveIntervalForTest ¶
SetKeepAliveIntervalForTest changes the keep-alive cadence for the duration of a single test. Registers t.Cleanup to restore the original value, so tests can't accidentally bleed configuration between cases. Intended for use only from _test.go files.
The new cadence only affects Hub instances constructed AFTER the call: runKeepAlive captures the duration in its time.Ticker at hub startup, so changing it later doesn't retro-actively shorten an already-running ticker. Build the hub inside the test (or right after SetKeepAliveIntervalForTest) to apply the override.
Types ¶
type Event ¶
type Event struct {
Type string
ID string
Payload json.RawMessage
}
Event is the wire shape the control plane pushes into the hub. Type is the CloudEvent discriminator (e.g. "user.created") and drives the SSE `event:` field. ID is the SSE `id:` field — the offset cursor a consumer resumes from; when empty the message is delivered live but never stored for replay (error control frames). Payload is the raw JSON body streamed in the SSE `data:` field — it MUST include the same `type` value (the OpenAPI schema requires it as the oneOf discriminator), but this type doesn't enforce that; the /admin0/events handler validates against the schema before calling Hub.Publish.
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub is the SSE fan-out the mock owns. One Hub per process; the HTTP handler at GET /events is hub.Handler(), and POST /admin0/events pushes via hub.Publish. Hub is safe for concurrent use; every underlying primitive (sse.Server, sse.Joe, recordingReplayer) is.
Lifecycle:
- NewHub starts a keep-alive goroutine.
- Reset drains current subscribers and rebuilds the underlying server + replay buffer, so /admin0/reset between tests is non-destructive to the hub itself.
- Shutdown drains every subscriber permanently and stops the keep-alive goroutine; intended for process shutdown.
func NewHub ¶
NewHub constructs a Hub. BufferSize is the cap of the replay buffer (used for resume via Last-Event-ID / ?from / ?from_timestamp); values <= 0 disable replay entirely (sse.Joe accepts a nil Replayer); values of 1 are clamped to 2 because the library requires a count of at least 2. Now is the clock the replayer's timestamp index uses; nil falls back to time.Now. The caller should wire this to internal/clock.Clock.Now when a controllable clock is present so ?from_timestamp behaves deterministically in clock-controlled tests. Opts customise the hub — e.g. WithReconnectHint sets the SSE retry: value sent on connect.
func (*Hub) ActiveSubscribers ¶ added in v0.229.0
ActiveSubscribers reports how many subscribers are connected to GET /events right now. A subscriber leaves the active set only when the server's read loop notices its connection closed, so a reading taken immediately after a client disconnects may briefly lag.
func (*Hub) Handler ¶
Handler returns the HTTP handler for GET /events. Wire it under bearer middleware at mount time.
The handler:
- Disables the http.Server WriteTimeout for this connection (SSE is long-lived; the server default would tear down healthy subscribers after the configured timeout).
- Promotes Auth0's ?from and ?from_timestamp query parameters to the SSE-spec Last-Event-ID header so the library's native replay path picks them up. ?from wins over ?from_timestamp. ?from_timestamp accepts RFC 3339; clients that send the timezone `+` unencoded (which URL-decodes to space) are tolerated by retrying with the space restored.
- Surfaces aged-out resume requests as 410 Gone (matching the OpenAPI declaration). Unparseable ?from_timestamp returns 400 with the standard mgmt error envelope.
- Pre-flushes the SSE response headers so http.Client.Do returns immediately rather than waiting for the first event.
- Tracks the request context in the active set so Reset / Shutdown can drain in-flight subscribers cleanly.
- Delegates to the underlying *sse.Server, which uses an OnSession callback to parse `?event_type=...` into the subscriber's topic list.
func (*Hub) Publish ¶
Publish broadcasts evt to every subscriber whose topic set intersects. A regular event is sent to broadcastTopic (reaches every filterless subscriber) and to evt.Type (reaches every filtered subscriber that listed this type). Error frames and offset-only progress markers are stream-wide control signals routed to the every-subscriber topic — see the branches below. Keep-alives use a separate topic and never go through this method.
The RLock is held across server.Publish so a concurrent Reset can't swap h.server underneath an in-flight publish and produce a spurious "provider is closed" error.
func (*Hub) Reset ¶
Reset swaps in a fresh server + replay buffer (so any concurrent Publish atomically moves to the new instance), drains the subscribers that were attached to the old server, then shuts the old server down — all while the new server is already serving new subscribers and publishes. The swap-before-shutdown ordering is what closes the Publish/Reset race: a publisher that grabbed the RLock immediately before Reset's mu.Lock acquired sees the OLD server (and Reset's Lock blocks until the publish completes because Publish holds RLock across the call); every publisher after that sees the NEW server. The OLD server is then shut down with no concurrent publish in flight.
Intended for the /admin0/reset control-plane hook between tests. Idempotent under concurrent callers (serialised via lifecycleMu).
func (*Hub) Shutdown ¶
Shutdown drains every subscriber, stops the keep-alive goroutine, and marks the hub closed permanently. Intended for process shutdown. Idempotent — extra calls are no-ops.
Uses the same swap-before-shutdown ordering as Reset to keep in-flight publishers race-free: the swap to a nil server happens atomically under mu.Lock, then the old server is shut down with no lock held.
func (*Hub) TotalSubscribers ¶ added in v0.229.0
TotalSubscribers reports how many subscriptions have connected since the hub was created or last Reset. It increments on every connect and never decrements within a window; Reset zeroes it.
type HubOption ¶ added in v0.230.0
type HubOption func(*Hub)
HubOption customises a Hub at construction.
func WithReconnectHint ¶ added in v0.230.0
WithReconnectHint sets the SSE `retry:` reconnect-delay hint sent on connect; <= 0 omits it. Wire it to the EVENTS_RECONNECT_HINT config. Defaults to DefaultReconnectHint when unset.
Source Files