Documentation
¶
Overview ¶
Package events is the in-process pub/sub for presence, device, audit and hbbs health events, exposed to the panel as Server-Sent Events at /api/v1/events.
Stream format (text/event-stream):
retry: 5000
id: 42
event: presence.changed
data: {"id":"123456789","online":true,"since":"...","source":"hbbs"}
event: resync
data: {"dropped":17}
: keepalive
"resync" tells the client that events were dropped for it (slow consumer) and it should refetch state over the REST API. Last-Event-ID is not supported: on reconnect clients refetch state.
Optional query parameter types=presence.changed,hbbs.health restricts the stream to the given event types.
Index ¶
Constants ¶
const DefaultBuffer = 256
DefaultBuffer is the per-subscriber queue length.
const (
DefaultKeepalive = 15 * time.Second
)
Defaults for the SSE handler.
const RoleAdmin = "admin"
Role values understood by Principal.IsAdmin.
Variables ¶
var Admins = Audience{AdminOnly: true}
Admins is the audience of admin-only events.
var ErrUnauthenticated = errors.New("events: unauthenticated")
ErrUnauthenticated is the conventional Authenticator error for "no valid session".
Functions ¶
Types ¶
type Audience ¶
type Audience struct {
// AdminOnly events are delivered to admins only.
AdminOnly bool
// UserIDs, when non-empty, limits delivery of a non-admin event to these
// users (admins always receive it). Empty means every authenticated user.
UserIDs []int64
}
Audience restricts who may receive an event.
type Authenticator ¶
Authenticator resolves the panel session of a request (stream B implements it on top of the session cookie). It is called once when the stream opens and again on every keepalive, so a revoked session ends the stream.
var DenyAll Authenticator = AuthenticatorFunc(func(*http.Request) (Principal, error) { return Principal{}, ErrUnauthenticated })
DenyAll is the default Authenticator: every request is rejected.
type AuthenticatorFunc ¶
AuthenticatorFunc adapts a function to Authenticator.
func (AuthenticatorFunc) Authenticate ¶
func (f AuthenticatorFunc) Authenticate(r *http.Request) (Principal, error)
Authenticate implements Authenticator.
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus is an in-process fan-out pub/sub. Publish never blocks: each subscriber has a bounded queue and, when it is full, the oldest queued event is dropped and counted.
func (*Bus) Publish ¶
Publish stamps and delivers an event to every matching subscriber. A nil *Bus is a valid no-op publisher.
func (*Bus) Subscribe ¶
func (b *Bus) Subscribe(filter func(Event) bool, buffer int) *Subscription
Subscribe registers a subscriber. filter (may be nil) selects events; buffer <= 0 means DefaultBuffer. Call Close when done.
func (*Bus) Subscribers ¶
Subscribers returns the current number of subscribers.
type Deps ¶
type Deps struct {
Log *slog.Logger
Store *store.Store
Bus *Bus
// Auth authenticates the stream; nil means DenyAll.
Auth Authenticator
// Keepalive is the comment/heartbeat interval (default 15 s).
Keepalive time.Duration
// Buffer is the per-stream queue length (default DefaultBuffer).
Buffer int
}
Deps are the collaborators the SSE endpoint needs.
type DeviceUpdate ¶
DeviceUpdate is the payload of DeviceUpdated.
type Event ¶
type Event struct {
// ID is a process-local, monotonically increasing sequence number.
ID uint64 `json:"id"`
Type Type `json:"type"`
Time time.Time `json:"time"`
Data any `json:"data"`
Audience Audience `json:"-"`
}
Event is one published message.
type Health ¶
type Health struct {
// Component: "hbbs" (rendezvous port, presence poller), "hbbs-admin" or "hbbr-admin".
Component string `json:"component"`
Up bool `json:"up"`
// RTTMillis is the round-trip time of the probe that observed the state.
RTTMillis float64 `json:"rtt_ms,omitempty"`
Error string `json:"error,omitempty"`
}
Health is the payload of HBBSHealth.
type Presence ¶
type Presence struct {
ID string `json:"id"`
Online bool `json:"online"`
// Since is when the device entered its current state (as observed by the panel).
Since time.Time `json:"since"`
// LastSeen is the last time the device was observed online (zero if never).
LastSeen time.Time `json:"last_seen,omitempty"`
// Source is "hbbs" (OnlineRequest) or "heartbeat" (fallback while hbbs is down).
Source string `json:"source"`
}
Presence is the payload of PresenceChanged.
type RelayState ¶ added in v0.3.0
type RelayState struct {
ID int64 `json:"id"`
Name string `json:"name"`
Health string `json:"health"`
Previous string `json:"previous,omitempty"`
Connected bool `json:"connected"`
Error string `json:"error,omitempty"`
}
RelayState is the payload of RelayHealth (internal/relays).
type Subscription ¶
type Subscription struct {
// contains filtered or unexported fields
}
Subscription is one consumer's bounded queue (a ring buffer).
func (*Subscription) Close ¶
func (s *Subscription) Close()
Close unregisters the subscription. It is idempotent.
func (*Subscription) Drain ¶
func (s *Subscription) Drain(dst []Event) ([]Event, uint64)
Drain appends all queued events to dst and returns it together with the number of events dropped since the previous Drain.
func (*Subscription) Dropped ¶
func (s *Subscription) Dropped() uint64
Dropped returns the total number of events dropped for this subscriber.
func (*Subscription) Next ¶
func (s *Subscription) Next(ctx context.Context) (Event, error)
Next blocks until an event is available or ctx ends.
func (*Subscription) Ready ¶
func (s *Subscription) Ready() <-chan struct{}
Ready is signalled (level-triggered, coalesced) when events may be queued.
type Type ¶
type Type string
Type names an event kind. It is used as the SSE "event:" field.
const ( // PresenceChanged: a device went online or offline (data: Presence). PresenceChanged Type = "presence.changed" // DeviceUpdated: a device row changed (data: DeviceUpdate). DeviceUpdated Type = "device.updated" // AuditConnection: a connection audit row was recorded (data: free-form object). AuditConnection Type = "audit.connection" // HBBSHealth: reachability of hbbs/hbbr changed (data: Health). HBBSHealth Type = "hbbs.health" // RelayHealth: a relay's health, agent connection or config state // changed (data: RelayState). Admins only. RelayHealth Type = "relay.health" )
Event kinds published by the panel.
Source Files