process

Documentation

Overview

Package process provides engine components that orchestrate the execution of process message handlers.

Index

• func InitializeHandler(ctx context.Context, db *sql.DB, handlerConfig *config.Process) error
• type DeadlinePump
  • func (p *DeadlinePump) AcquireDelivery(ctx context.Context, tx *sql.Tx) (messagepump.Delivery, bool, error)
  • func (p *DeadlinePump) HandleDelivery(ctx context.Context, tx *sql.Tx, delivery messagepump.Delivery, ...) error
  • func (*DeadlinePump) PostponeDelivery(ctx context.Context, tx *sql.Tx, delivery messagepump.Delivery, ...) error
• type EventPump
  • func (p *EventPump) AcquireDelivery(ctx context.Context, tx *sql.Tx) (messagepump.Delivery, bool, error)
  • func (p *EventPump) HandleDelivery(ctx context.Context, tx *sql.Tx, delivery messagepump.Delivery, ...) error
  • func (p *EventPump) PostponeDelivery(ctx context.Context, tx *sql.Tx, delivery messagepump.Delivery, ...) error

Functions

func InitializeHandler

func InitializeHandler(
	ctx context.Context,
	db *sql.DB,
	handlerConfig *config.Process,
) error

InitializeHandler initializes the state of a process message handler.

It is called by the engine on startup, before any components are started.

Types

type DeadlinePump

type DeadlinePump struct {
	DB                   *sql.DB
	Handler              dogma.ProcessMessageHandler[dogma.ProcessRoot]
	Identity             *identitypb.Identity
	Packer               *envelopepb.Packer
	DeadlineTypeIDs      *uuidpb.Set
	OutboundMessageTypes map[reflect.Type]struct{}
}

DeadlinePump is a messagepump.Driver that delivers pending deadlines to a process message handler.

func (*DeadlinePump) AcquireDelivery

func (p *DeadlinePump) AcquireDelivery(
	ctx context.Context,
	tx *sql.Tx,
) (messagepump.Delivery, bool, error)

AcquireDelivery attempts to acquire the next pending deadline for the handler.

func (*DeadlinePump) HandleDelivery

func (p *DeadlinePump) HandleDelivery(
	ctx context.Context,
	tx *sql.Tx,
	delivery messagepump.Delivery,
	envelope *envelopepb.Envelope,
	logger *slog.Logger,
) error

HandleDelivery dispatches a deadline to the process handler.

func (*DeadlinePump) PostponeDelivery

func (*DeadlinePump) PostponeDelivery(
	ctx context.Context,
	tx *sql.Tx,
	delivery messagepump.Delivery,
	delay time.Duration,
) error

PostponeDelivery reschedules the deadline for redelivery after delay.

type EventPump

type EventPump struct {
	DB                   *sql.DB
	Handler              dogma.ProcessMessageHandler[dogma.ProcessRoot]
	Identity             *identitypb.Identity
	Packer               *envelopepb.Packer
	EventTypeIDs         *uuidpb.Set
	OutboundMessageTypes map[reflect.Type]struct{}
	Logger               *slog.Logger
}

EventPump is a messagepump.Driver that delivers pending events to a process message handler.

func (*EventPump) AcquireDelivery

func (p *EventPump) AcquireDelivery(
	ctx context.Context,
	tx *sql.Tx,
) (messagepump.Delivery, bool, error)

AcquireDelivery attempts to acquire the next pending event for the handler on one of its tracked event streams.

If there are no relevant events available on the chosen stream, it advances the checkpoint offset to the end of the stream so that the stream is not re-acquired until new events arrive.

func (*EventPump) HandleDelivery

func (p *EventPump) HandleDelivery(
	ctx context.Context,
	tx *sql.Tx,
	delivery messagepump.Delivery,
	envelope *envelopepb.Envelope,
	logger *slog.Logger,
) error

HandleDelivery dispatches an event to the process handler.

func (*EventPump) PostponeDelivery

func (p *EventPump) PostponeDelivery(
	ctx context.Context,
	tx *sql.Tx,
	delivery messagepump.Delivery,
	delay time.Duration,
) error

PostponeDelivery reschedules consumption of the stream after delay.