Documentation
¶
Overview ¶
Package stream is the in-process live fan-out bus (docs/ROADMAP.md §7.2 file 6.13, docs/DESIGN.md §1.1): the engine publishes every committed data operation and device lifecycle event, and consumers — the M7b WebSocket/SSE endpoint, tests — subscribe per realm with optional filters. Publishing never blocks: a slow consumer's full channel drops the event for that consumer with a metric (§1.4 philosophy — live viewers must never backpressure ingestion).
Index ¶
Constants ¶
const ( // KindIncomingData is a committed device data operation (set or unset). KindIncomingData = "incoming_data" // KindDeviceConnected is a device connection. KindDeviceConnected = "device_connected" // KindDeviceDisconnected is a device disconnection. KindDeviceDisconnected = "device_disconnected" )
Event kinds mirror the Astarte trigger event names.
const DefaultSubscriberBuffer = 64
DefaultSubscriberBuffer is the per-subscriber channel capacity used when Subscribe is called with a non-positive buffer.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus is the fan-out hub. The zero value is not usable; construct with New.
func New ¶
func New(reg prometheus.Registerer) *Bus
New builds a bus; a non-nil reg receives its collectors.
func (*Bus) Close ¶
func (b *Bus) Close()
Close shuts the bus down: every subscriber channel closes and later publishes are no-ops.
func (*Bus) Publish ¶
Publish fans an event out to every matching subscriber without blocking; full subscriber channels drop the event with a metric.
func (*Bus) Subscribe ¶
Subscribe registers a consumer for one realm's events. The returned cancel function unregisters it and closes the channel; the channel also closes when the bus shuts down. buffer <= 0 selects DefaultSubscriberBuffer. Subscribing to a closed bus returns an already-closed channel.
func (*Bus) Subscribers ¶
Subscribers reports the number of registered consumers (metrics, tests).
type Event ¶
type Event struct {
// Kind discriminates the event (Kind* constants).
Kind string
// Realm is the tenant.
Realm string
// DeviceID is the encoded device ID.
DeviceID string
// Interface and Path locate data events; empty for lifecycle events.
Interface string
Path string
// Value is the JSON-friendly rendering of a data event's value (nil for
// property unset and lifecycle events).
Value any
// Timestamp is the event instant (the effective sample timestamp for
// data events).
Timestamp time.Time
}
Event is one live event.
type Filter ¶
type Filter struct {
// DeviceID keeps only one device's events when set.
DeviceID string
// Interface keeps only one interface's data events when set (lifecycle
// events carry no interface and are kept only by an empty filter).
Interface string
}
Filter narrows a subscription; zero-value fields match everything.
Source Files