observerruntime

Documentation

Overview

Package observerruntime owns ordered observer delivery and acknowledgement.

Index

• func DecodeContent(descriptor events.Descriptor, ec events.Context, content []byte, ...) (events.Context, any, error)
• func ReactorInvocationContext(ctx context.Context, id reactors.ID, ec events.Context) context.Context
• type Reactor
  • func Open(ctx context.Context, conn grpc.ClientConnInterface, connectionID string, ...) (*Reactor, error)
  • func (r *Reactor) Run(ctx context.Context) error
• type Reducer
  • func OpenReducer(ctx context.Context, conn grpc.ClientConnInterface, connectionID string, ...) (*Reducer, error)
  • func (r *Reducer) Run(ctx context.Context) error

Functions

func DecodeContent

func DecodeContent(descriptor events.Descriptor, ec events.Context, content []byte, generations map[events.Generation]json.RawMessage) (events.Context, any, error)

DecodeContent applies C#'s generation selection, falling back to Content when no requested representation is supplied. It never treats JSON null as an event.

func ReactorInvocationContext

func ReactorInvocationContext(ctx context.Context, id reactors.ID, ec events.Context) context.Context

ReactorInvocationContext applies the production identity and causation scope. Scenario adapters share it rather than defining test-only ambient semantics.

Types

type Reactor

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

Reactor owns a single duplex stream. Run has one receiver and one sender with no application queue: transport backpressure bounds pending messages.

func Open

func Open(ctx context.Context, conn grpc.ClientConnInterface, connectionID string, store metadata.StoreName, namespace metadata.Namespace, plan *reactors.Plan, runtime reactors.Runtime) (*Reactor, error)

Open sends the registration on an explicit generation lifetime. The protocol has no registration acknowledgement: success means sent, not kernel acceptance.

func (*Reactor) Run

func (r *Reactor) Run(ctx context.Context) error

Run processes batches sequentially until cancellation or stream failure. Effects are never retried here. The kernel controls failure recovery and resume position.

type Reducer

type Reducer struct {

	// OnChange receives successful local folds before kernel acknowledgement.
	// Install before Run. It must not block or call user code.
	OnChange func(string, json.RawMessage)
	// contains filtered or unexported fields
}

Reducer is the reducer protocol adapter over the shared ordered stream loop.

func OpenReducer

func OpenReducer(ctx context.Context, conn grpc.ClientConnInterface, connectionID string, store metadata.StoreName, namespace metadata.Namespace, plan *reducers.Plan) (*Reducer, error)

OpenReducer sends the immutable registration snapshot. Like reactors, the duplex protocol has no registration acknowledgement beyond a successful send.

func (*Reducer) Run

func (r *Reducer) Run(ctx context.Context) error

Run processes one complete fold and cleanup before acknowledging it. Replay notifications run serially in fresh scopes and have no result handshake.