Versions in this module Expand all Collapse all v1 v1.0.1 Aug 28, 2026 v1.0.0 Aug 26, 2026 Changes in this version + var ErrInvalidConfiguration = errors.New("valkeystream: invalid configuration") + var ErrInvalidDeliveryAttemptLimit = errors.New("valkeystream: invalid delivery attempt limit") + var ErrInvalidManagementStatus = errors.New("valkeystream: invalid management status") + var ErrManagementControlDisabled = errors.New("valkeystream: management control disabled") + var ErrManagementRecordNotFound = fmt.Errorf("valkeystream: management record not found: %w", management.ErrRecordNotFound) + var ErrManagementRecordsDisabled = fmt.Errorf("valkeystream: management records disabled: %w", ...) + var ErrManagementStatusDisabled = errors.New("valkeystream: management status disabled") + type ConfigurationError struct + Cause error + Field string + func (e *ConfigurationError) Error() string + func (e *ConfigurationError) Unwrap() []error + type DeliveryAttemptLimitResolver func(core.TaskMessage) int64 + type Option func(*options) error + func WithAddress(address string) Option + func WithAuthentication(username, password string) Option + func WithBlockTime(timeout time.Duration) Option + func WithBlockingPool(minimum, maximum int, cleanup time.Duration) Option + func WithCanceledDeadLetterCodes(codes ...string) Option + func WithClientName(name string) Option + func WithCommandTimeout(timeout time.Duration) Option + func WithConsumer(name string) Option + func WithDB(database int) Option + func WithDeadLetter(stream string, maxAttempts int64) Option + func WithDeliveryAttemptLimitResolver(resolver DeliveryAttemptLimitResolver) Option + func WithDialTimeout(timeout time.Duration) Option + func WithFailureStream(stream string) Option + func WithGroup(name string) Option + func WithLogger(logger queue.Logger) Option + func WithManagementStatus(metadata management.StatusMetadata) Option + func WithMaxLength(length int64) Option + func WithReadBatchSize(size int) Option + func WithReclaim(minIdle, interval time.Duration, batchSize int) Option + func WithRecordRetention(maxRecords int64) Option + func WithReplayDestinations(destinations ...string) Option + func WithRequestTimeout(timeout time.Duration) Option + func WithRunFunc(run func(context.Context, core.TaskMessage) error) Option + func WithShutdownTimeout(timeout time.Duration) Option + func WithStreamName(name string) Option + func WithTLSConfig(config *tls.Config) Option + type Publisher struct + func NewPublisherE(option ...Option) (*Publisher, error) + func (*Publisher) BackendName() string + func (publisher *Publisher) Queue(message core.QueuedMessage, options ...job.AllowOption) error + func (publisher *Publisher) QueueName() string + func (publisher *Publisher) Shutdown() error + type Stats struct + Acknowledged uint64 + DeadLettered uint64 + Delivered uint64 + Depth int64 + Enqueued uint64 + Lag int64 + LagKnown bool + OldestPendingAge time.Duration + Pending int64 + Reclaimed uint64 + Retries uint64 + SettlementFailures uint64 + type Worker struct + func NewWorker(option ...Option) *Worker + func NewWorkerE(option ...Option) (*Worker, error) + func (*Worker) BackendName() string + func (w *Worker) Execute(ctx context.Context, command management.Command) (management.CommandResult, error) + func (w *Worker) Inspect(ctx context.Context, request management.InspectRequest) (management.JobRecord, error) + func (w *Worker) ListDeadLetters(ctx context.Context, request management.PageRequest) (management.RecordPage, error) + func (w *Worker) ListFailures(ctx context.Context, request management.PageRequest) (management.RecordPage, error) + func (w *Worker) ObserveQueue(ctx context.Context) (management.QueueStatus, error) + func (w *Worker) ObserveWorker(ctx context.Context) (management.WorkerStatus, error) + func (w *Worker) Queue(task core.TaskMessage) error + func (w *Worker) QueueName() string + func (w *Worker) Request() (core.TaskMessage, error) + func (w *Worker) Run(ctx context.Context, task core.TaskMessage) error + func (w *Worker) Shutdown() error + func (w *Worker) Stats(ctx context.Context) (Stats, error)
Go to main page