sse

sse

import "github.com/agentstation/starmap/internal/server/sse"

Package sse provides the sole reactive catalog-publication transport.

Index

• Constants
• type Broadcaster
  • func NewBroadcaster(config Config, logger *zerolog.Logger) (*Broadcaster, error)
  • func (b *Broadcaster) ClientCount() int
  • func (b *Broadcaster) Close()
  • func (b *Broadcaster) Health() Health
  • func (b *Broadcaster) Publish(publication Publication) error
  • func (b *Broadcaster) ServeHTTP(writer http.ResponseWriter, request *http.Request)
  • func (b *Broadcaster) Stats() DeliveryStats
• type Config
• type DeliveryError
• type DeliveryStats
• type Health
• type Publication
• type StreamState

Constants

const (
    // CatalogPublishedEvent is the stable SSE event name.
    CatalogPublishedEvent = remote.CatalogPublishedEvent
    // DefaultHeartbeatInterval keeps idle streams alive through common proxies.
    DefaultHeartbeatInterval = 20 * time.Second
    // DefaultWriteTimeout bounds each event or heartbeat write and flush. The
    // broadcaster resets the deadline before every frame, so the value bounds
    // one frame and never bounds the stream. Two minutes lets a slow reader on
    // a congested link keep its subscription.
    DefaultWriteTimeout = 2 * time.Minute

    // DefaultMaxClients bounds concurrent subscriber connections. Each
    // connection owns one writer goroutine and one queued publication, so an
    // unbounded client count would let subscribers exhaust publisher memory.
    DefaultMaxClients = 512

    // DefaultAdmissionRetryAfter is the base wait a refused subscriber gets.
    // The served value carries jitter, so a refused fleet does not return at
    // one instant.
    DefaultAdmissionRetryAfter = 30 * time.Second
)

type Broadcaster

Broadcaster delivers publications to HTTP SSE connections. Each connection owns exactly one writer goroutine: its request handler. Publication overload terminates that connection so reconnect catch-up can recover without a silently healthy stream.

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

func NewBroadcaster

func NewBroadcaster(config Config, logger *zerolog.Logger) (*Broadcaster, error)

NewBroadcaster constructs an idle broadcaster. It starts no goroutine.

func (*Broadcaster) ClientCount

func (b *Broadcaster) ClientCount() int

ClientCount returns the number of currently registered SSE connections.

func (*Broadcaster) Close

func (b *Broadcaster) Close()

Close terminates every active connection and rejects new ones.

func (*Broadcaster) Health

func (b *Broadcaster) Health() Health

Health returns the current server-side stream delivery health.

func (*Broadcaster) Publish

func (b *Broadcaster) Publish(publication Publication) error

Publish offers one committed generation to every connected stream. It never blocks the catalog commit path. Publish terminates a connection that cannot accept the generation, which prevents silent data loss.

func (*Broadcaster) ServeHTTP

func (b *Broadcaster) ServeHTTP(writer http.ResponseWriter, request *http.Request)

ServeHTTP serves one heartbeat-enabled SSE connection.

func (*Broadcaster) Stats

func (b *Broadcaster) Stats() DeliveryStats

Stats returns cumulative delivery counters.

type Config

Config controls per-connection SSE liveness and write behavior.

type Config struct {
    HeartbeatInterval time.Duration
    WriteTimeout      time.Duration

    // MaxClients bounds concurrent subscriber connections. Zero selects
    // DefaultMaxClients, and a negative value is invalid.
    MaxClients int

    // AdmissionRetryAfter is the base wait the publisher declares to a refused
    // subscriber. Zero selects DefaultAdmissionRetryAfter.
    AdmissionRetryAfter time.Duration
}

type DeliveryError

DeliveryError classifies the latest stream failure without exposing secrets.

type DeliveryError struct {
    Kind       string    `json:"kind"`
    OccurredAt time.Time `json:"occurred_at"`
}

type DeliveryStats

DeliveryStats is a lock-free snapshot of SSE delivery behavior.

type DeliveryStats struct {
    Published              uint64 `json:"published"`
    Sent                   uint64 `json:"sent"`
    Heartbeats             uint64 `json:"heartbeats"`
    Disconnected           uint64 `json:"disconnected"`
    Refused                uint64 `json:"refused"`
    BackpressureTerminated uint64 `json:"backpressure_terminated"`
    Failed                 uint64 `json:"failed"`
}

type Health

Health reports server-side SSE publication delivery without conflating heartbeat liveness with catalog freshness.

type Health struct {
    State            StreamState    `json:"state"`
    Clients          int            `json:"clients"`
    MaxClients       int            `json:"max_clients"`
    LastHeartbeatAt  time.Time      `json:"last_heartbeat_at"`
    LastEventAt      time.Time      `json:"last_event_at"`
    LastGenerationID string         `json:"last_generation_id,omitempty"`
    LastSequence     uint64         `json:"last_sequence"`
    LastError        *DeliveryError `json:"last_error,omitempty"`
    Delivery         DeliveryStats  `json:"delivery"`
}

type Publication

Publication identifies one committed immutable catalog generation.

type Publication = remote.Publication

type StreamState

StreamState is the server-side publication stream state.

type StreamState string

const (
    // StreamStateIdle means the broadcaster accepts streams but has no clients.
    StreamStateIdle StreamState = "idle"
    // StreamStateStreaming means the broadcaster has active clients.
    StreamStateStreaming StreamState = "streaming"
    // StreamStateStopped means the broadcaster rejects new streams.
    StreamStateStopped StreamState = "stopped"
)

Generated by gomarkdoc