Documentation
¶
Overview ¶
Package messagepump generalizes a message pump that acquires messages for processing and dispatches them to a worker pool.
Index ¶
- Variables
- func AdvanceStreamCheckpoint(ctx context.Context, tx *sql.Tx, logger *slog.Logger, ...) error
- func PostponeCommandDelivery(ctx context.Context, tx *sql.Tx, delivery Delivery, delay time.Duration) error
- func PostponeEventDelivery(ctx context.Context, tx *sql.Tx, handlerKey *uuidpb.UUID, delivery Delivery, ...) error
- func SetCheckpointOffset(ctx context.Context, tx *sql.Tx, handlerKey, streamID *uuidpb.UUID, ...) error
- type Delivery
- type DeliveryStream
- type Driver
- type MessagePump
Constants ¶
This section is empty.
Variables ¶
var ErrBusy = errors.New("delivery target is busy")
ErrBusy is returned by Driver.HandleDelivery to indicate that the delivery target is temporarily unavailable (e.g. locked by another transaction) and should be retried.
var ErrFailed = errors.New("delivery failed")
ErrFailed is returned by Driver.HandleDelivery to indicate that the delivery failed and should be postponed for redelivery after an exponential backoff.
Functions ¶
func AdvanceStreamCheckpoint ¶
func AdvanceStreamCheckpoint( ctx context.Context, tx *sql.Tx, logger *slog.Logger, handlerKey, streamID *uuidpb.UUID, oldCheckpointOffset, newCheckpointOffset uint64, ) error
AdvanceStreamCheckpoint updates the checkpoint offset of a handler's row in eventstream.handler_checkpoints and resets its failure counter.
func PostponeCommandDelivery ¶
func PostponeCommandDelivery( ctx context.Context, tx *sql.Tx, delivery Delivery, delay time.Duration, ) error
PostponeCommandDelivery postpones a queued command for redelivery after delay.
func PostponeEventDelivery ¶
func PostponeEventDelivery( ctx context.Context, tx *sql.Tx, handlerKey *uuidpb.UUID, delivery Delivery, delay time.Duration, ) error
PostponeEventDelivery reschedules consumption of a stream after delay, and sets the failure count to the given value.
func SetCheckpointOffset ¶
func SetCheckpointOffset( ctx context.Context, tx *sql.Tx, handlerKey, streamID *uuidpb.UUID, oldCheckpointOffset, newCheckpointOffset uint64, ) error
SetCheckpointOffset sets the checkpoint offset of a handler's row in eventstream.handler_checkpoints and resets its failure counter.
Unlike AdvanceStreamCheckpoint, the new offset need not be greater than the old offset. It is used when the new offset is supplied by the handler itself (via optimistic concurrency control), which may set it to any value.
Types ¶
type Delivery ¶
type Delivery struct {
// MessageID is the unique identifier of the message being delivered.
MessageID *uuidpb.UUID
// MessageTypeID is the unique identifier of the message's type.
MessageTypeID *uuidpb.UUID
// EnvelopeBytes is the serialized form of the message's envelope.
EnvelopeBytes []byte
// Failures is the number of times that delivery has been attempted and
// failed.
Failures uint64
// Stream describes the position of the message within an event stream. It
// is set by stream-based pumps and is nil for queue-based pumps.
Stream *DeliveryStream
}
Delivery describes a unit of work scoped to the delivery of a single message to a single handler.
func AcquireCommandDelivery ¶
func AcquireCommandDelivery( ctx context.Context, tx *sql.Tx, messageTypeIDs *uuidpb.Set, ) (Delivery, bool, error)
AcquireCommandDelivery attempts to acquire the next pending command for the handler from the command queue.
func AcquireEventDelivery ¶
func AcquireEventDelivery( ctx context.Context, tx *sql.Tx, logger *slog.Logger, handlerKey *uuidpb.UUID, messageTypeIDs *uuidpb.Set, ) (Delivery, bool, error)
AcquireEventDelivery attempts to acquire the next pending event for the handler.
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.
type DeliveryStream ¶
type DeliveryStream struct {
// ID is the unique identifier of the stream.
ID *uuidpb.UUID
// EventOffset is the offset of the message within the stream.
EventOffset uint64
// CheckpointOffset is the engine's recorded checkpoint offset for the
// stream at the moment the delivery was acquired.
CheckpointOffset uint64
}
DeliveryStream describes the position of a message within an event stream, and the engine's recorded checkpoint offset for that stream at the moment the delivery was acquired.
type Driver ¶
type Driver interface {
// AcquireDelivery attempts to acquire the next pending [Delivery] within
// tx.
//
// If there are no pending deliveries, ok is false.
AcquireDelivery(ctx context.Context, tx *sql.Tx) (del Delivery, ok bool, err error)
// PostponeDelivery schedules redelivery after the specified delay.
//
// The delivery's failure count may be greater than the value returned by
// AcquireDelivery, indicating that the postponement is due to a failure.
PostponeDelivery(
ctx context.Context,
tx *sql.Tx,
del Delivery,
delay time.Duration,
) error
// HandleDelivery processes a [Delivery] within tx.
//
// It returns [ErrFailed] to indicate that the delivery should be postponed
// for redelivery with an incremented failure count, or [ErrBusy] to
// indicate that the delivery should be retried without incrementing the
// failure count.
HandleDelivery(
ctx context.Context,
tx *sql.Tx,
del Delivery,
envelope *envelopepb.Envelope,
logger *slog.Logger,
) error
}
Driver implements the underlying messaging operations orchestrated by a MessagePump.
type MessagePump ¶
type MessagePump struct {
Driver Driver
DB *sql.DB
Workers int
PollInterval time.Duration
BackoffBase, BackoffCap time.Duration
Logger *slog.Logger
}
MessagePump is a helper for implementing a message pump that acquires and handles deliveries concurrently.
func (*MessagePump) Run ¶
func (p *MessagePump) Run(ctx context.Context)
Run runs the message pump until ctx is canceled.
Source Files