stream

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
• type Bus
  • func New(reg prometheus.Registerer) *Bus
  • func (b *Bus) Close()
  • func (b *Bus) Publish(ev Event)
  • func (b *Bus) Subscribe(realm string, f Filter, buffer int) (<-chan Event, func())
  • func (b *Bus) Subscribers() int
• type Event
• type Filter

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.

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

func (b *Bus) Publish(ev Event)

Publish fans an event out to every matching subscriber without blocking; full subscriber channels drop the event with a metric.

func (*Bus) Subscribe

func (b *Bus) Subscribe(realm string, f Filter, buffer int) (<-chan Event, func())

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

func (b *Bus) Subscribers() int

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.