eventstream

Documentation

Overview

Package eventstream is a single-process workflow.EventStreamer backed by the sqlite event_log / event_cursors tables (see internal/store), so an in-flight event survives a daemon restart. Recv parks on a sync.Cond instead of busy-spinning — see ADR-0033.

The default in-memory adapter shipped with luno/workflow (adapters/memstreamer) implements Recv as an unconditional `for { ... continue }` loop with no backoff: when the log is exhausted, every consumer goroutine hammers two mutexes in a tight cycle. With ~7 step consumers + a timeout consumer in everflow's refactor-sweep, that pegs all cores at ~380% CPU even when there's nothing to do.

Semantics match memstreamer: topic filter, per-receiver cursor, StreamFromLatest option. Two behavioural differences: Recv blocks on cond.Wait until a Send broadcasts (or ctx is cancelled), and the log + cursors live in sqlite rather than an in-process slice/map, so a daemon restart resumes every receiver from its last acked position instead of replaying (or losing) events. See ADR-0049.

Index

• type Streamer
  • func New(db *sql.DB) *Streamer
  • func (s *Streamer) NewReceiver(ctx context.Context, topic string, name string, ...) (workflow.EventReceiver, error)
  • func (s *Streamer) NewSender(_ context.Context, topic string) (workflow.EventSender, error)

Types

type Streamer

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

Streamer fans out workflow events to per-step receivers. Safe for concurrent senders and receivers; one Streamer per daemon process.

func New

func New(db *sql.DB) *Streamer

New returns a Streamer ready to wire into workflow.Build's EventStreamer slot. db must already have the event_log / event_cursors schema applied (internal/store.OpenSqlite does this) and is typically the same handle backing the daemon's RecordStore / TimeoutStore.

func (*Streamer) NewReceiver

func (s *Streamer) NewReceiver(
	ctx context.Context,
	topic string,
	name string,
	opts ...workflow.ReceiverOption,
) (workflow.EventReceiver, error)

func (*Streamer) NewSender

func (s *Streamer) NewSender(_ context.Context, topic string) (workflow.EventSender, error)