Versions in this module Expand all Collapse all v0 v0.1.0 Jul 1, 2026 Changes in this version + const DefaultBatchMaxRows + const DefaultBatchMaxWait + const DefaultShardQueue + const DefaultShards + var ErrInterfaceNotFound = errors.New("engine: interface not installed") + var ErrNotAProperty = errors.New("engine: interface is not a properties interface") + var ErrNotServerOwned = errors.New("engine: interface is not server-owned") + var ErrPathNotFound = errors.New("engine: path matches no endpoint") + var ErrRealmUnknown = errors.New("engine: realm unknown") + var ErrUnsetNotAllowed = errors.New("engine: mapping does not allow unset") + type BrokerPort interface + Publish func(topic string, payload []byte, qos byte, retain bool, expiry time.Duration) error + RefreshIntrospection func(ctx context.Context, realm string, id deviceid.ID) error + func AdaptBroker(b *broker.Broker) BrokerPort + type Config struct + BatchMaxRows int + BatchMaxWait time.Duration + Logger *slog.Logger + MaxPayloadBytes int + Registerer prometheus.Registerer + ShardQueue int + Shards int + type Engine struct + func New(st Store, bp BrokerPort, cfg Config) (*Engine, error) + func (e *Engine) AttachBroker(bp BrokerPort) + func (e *Engine) Bus() *stream.Bus + func (e *Engine) Drain(ctx context.Context) error + func (e *Engine) OnLifecycleEvent(ev broker.LifecycleEvent) + func (e *Engine) PublishServerValue(ctx context.Context, realm string, id deviceid.ID, ifaceName, path string, ...) error + func (e *Engine) RefreshInterfaces(ctx context.Context, realmID int16) error + func (e *Engine) RefreshTriggers(ctx context.Context, realmID int16) error + func (e *Engine) Start(ctx context.Context) error + func (e *Engine) Submit(m broker.InboundMessage) + func (e *Engine) UnsetServerProperty(ctx context.Context, realm string, id deviceid.ID, ifaceName, path string) error + type OpKind uint8 + const OpIndividual + const OpObject + const OpPropertySet + const OpPropertyUnset + func (k OpKind) String() string + type PersistOp struct + DeviceID deviceid.ID + Format payload.Format + Interface *interfaceschema.CompiledInterface + Kind OpKind + Mapping *interfaceschema.CompiledMapping + Path string + Realm string + RealmID int16 + ReceptionTS time.Time + TS time.Time + Value payload.Value + type Store interface + AppendDatastreams func(ctx context.Context, batch store.DatastreamBatch) error + GetDevice func(ctx context.Context, realmID int16, id deviceid.ID) (*store.Device, error) + ListRealms func(ctx context.Context) ([]store.Realm, error) + ListServerOwnedProperties func(ctx context.Context, realmID int16, deviceID deviceid.ID) ([]store.Property, error) + ListTriggers func(ctx context.Context, realmID int16) ([]store.Trigger, error) + Listen func(ctx context.Context, channel string) (<-chan store.Notification, error) + LoadRealmInterfaces func(ctx context.Context, realmID int16) ([]*store.StoredInterface, error) + PurgeDeviceOwnedExcept func(ctx context.Context, realmID int16, deviceID deviceid.ID, ...) (int64, error) + SetPayloadFormatHint func(ctx context.Context, realmID int16, id deviceid.ID, hint string) error + UnsetProperty func(ctx context.Context, realmID int16, deviceID deviceid.ID, interfaceID int64, ...) (bool, error) + UpdateIntrospection func(ctx context.Context, realmID int16, id deviceid.ID, ...) (map[string]store.InterfaceVersion, error) + UpsertProperty func(ctx context.Context, p store.Property) error
Go to main page