aggregate

package

v0.1.2 Latest Latest Go to latest Published: Jul 4, 2026 License: MIT Imports: 17 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/dogmatiq/runkit

Links

Open Source Insights

Documentation ¶

Overview ¶

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

Index ¶

type CommandPump

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type CommandPump ¶

type CommandPump struct {
	DB                   *sql.DB
	Handler              dogma.AggregateMessageHandler[dogma.AggregateRoot]
	Identity             *identitypb.Identity
	Packer               *envelopepb.Packer
	CommandTypeIDs       *uuidpb.Set
	OutboundMessageTypes map[reflect.Type]struct{}
}

CommandPump is a messagepump.Driver that delivers pending commands to an aggregate message handler of a specific type.

func (*CommandPump) AcquireDelivery ¶

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

AcquireDelivery attempts to acquire the next pending command for an aggregate handler of one of the configured types.

func (*CommandPump) HandleDelivery ¶

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

HandleDelivery dispatches a command to the aggregate handler.

func (*CommandPump) PostponeDelivery ¶

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

PostponeDelivery reschedules the command for redelivery after delay.

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL