streamruntime

type DeclStats ¶

type DeclStats struct {
	Streams    int
	Consumers  int
	Publishers int
}

DeclStats is the operator-facing count the unconfigured-URI error names.

type Declarations ¶

type Declarations interface {
	IsEmpty() bool
	Stats() DeclStats
}

Declarations is the collected, validated declaration set. The concrete type is produced by the registered runtime so app never names streams types.

type Handle ¶

type Handle interface {
	Start(ctx context.Context, decls Declarations) error
	Close() error
	StopConsumers()
	SetTenantStamps(enabled bool)
	Ready() bool
	Stats() map[string]any
}

Handle is the started (or about-to-start) stream manager as the framework sees it.

type HeldMessage ¶

type HeldMessage struct {
	Consumer   string
	Stream     string
	Offset     int64
	TenantID   string
	Data       []byte
	Properties map[string]any
	HeldAt     time.Time
}

HeldMessage is one parked stream delivery as the hold ledger sees it.

type HoldLedger ¶

type HoldLedger interface {
	Park(ctx context.Context, msg *HeldMessage) error
	HeldTenants(ctx context.Context, consumer string) ([]string, error)
}

HoldLedger is the port stream consumers park through.

Park is idempotent on (Consumer, Stream, Offset) and marks the tenant held in the same write: a row whose tenant is not held would be replayed by nothing.

type HoldReplayer ¶

type HoldReplayer interface {
	HoldConsumers() []string
	Replay(ctx context.Context, consumer string, msg *HeldMessage) error
	// ReloadHeld refreshes one consumer's held set from the ledger. It takes no
	// listing: the generation that guards the set has to be read BEFORE the ledger
	// is, and only the streams package holds it, so the read belongs on that side
	// of the port.
	ReloadHeld(ctx context.Context, consumer string) error
}

HoldReplayer is what the hold drain drives to put a held message back through the lane, and how it tells a runner which tenants the ledger still holds.

type ManagerOptions ¶

type ManagerOptions struct {
	URI                 string
	AddressResolverHost string
	AddressResolverPort int
	OffsetStoreCount    int
	OffsetStoreInterval time.Duration
	Logger              logger.Logger
	Hold                HoldLedger
}

ManagerOptions is the subset of manager construction the framework owns.

type ModuleNamer ¶ added in v0.63.0

type ModuleNamer interface {
	Name() string
}

ModuleNamer is the subset of app.Module the runtime needs to walk declarers.

type Runtime ¶

type Runtime interface {
	CollectDeclarations(modules []ModuleNamer, log logger.Logger) (Declarations, error)
	NewManager(opts *ManagerOptions) Handle
	CanDrainHold() bool
}

Runtime is the registered streams implementation.

func Registered() Runtime

func SwapRegistered(r Runtime) Runtime