queueprocessor

May 21, 2026

Changes in this version

+ type GoqueTypedProcessor struct

+ func (p *GoqueTypedProcessor[T]) ProcessTask(ctx context.Context, task *entity.Task) error

+ type GoqueTypedProcessorOpts func(*GoqueTypedProcessor[T])

+ func WithCancelTaskWhenPayloadDecodeError[T any]() GoqueTypedProcessorOpts[T]

type TaskProcessor

+ func NewTypedTaskProcessor[T any](processor TypedTaskProcessor[T], opts ...GoqueTypedProcessorOpts[T]) TaskProcessor

+ type TypedTaskProcessor interface

+ ProcessTask func(ctx context.Context, task *entity.TypedTask[T]) error

+ func NoopTypedTaskProcessor[T any]() TypedTaskProcessor[T]

+ type TypedTaskProcessorFunc func(ctx context.Context, task *entity.TypedTask[T]) error

+ func (f TypedTaskProcessorFunc[T]) ProcessTask(ctx context.Context, task *entity.TypedTask[T]) error

Mar 19, 2026

Changes in this version

type GoqueProcessorOpts

+ func WithDisableVerboseLogging() GoqueProcessorOpts

Nov 28, 2025

Changes in this version

+ var ErrTaskCancel = errors.New("task canceled")

+ var ErrTaskTimeout = errors.New("task processing timeout")

+ var FibonacciPeriods = []time.Duration

+ var StrongPeriods = []time.Duration

+ func LoggingAfterProcessing(ctx context.Context, task *entity.Task, err error)

+ func LoggingBeforeProcessing(ctx context.Context, task *entity.Task)

+ type GoqueProcessor struct

+ func NewGoqueProcessor(taskStorage TaskStorage, taskType entity.TaskType, processor TaskProcessor, ...) *GoqueProcessor

+ func (p *GoqueProcessor) Name() string

+ func (p *GoqueProcessor) Run(ctx context.Context) error

+ func (p *GoqueProcessor) Stop()

+ type GoqueProcessorOpts func(*GoqueProcessor)

+ func WithCleanerPeriod(period time.Duration) GoqueProcessorOpts

+ func WithCleanerTimeout(timeout time.Duration) GoqueProcessorOpts

+ func WithCleanerUpdatedAtTimeAgo(updatedAtTimeAgo time.Duration) GoqueProcessorOpts

+ func WithHealerPeriod(period time.Duration) GoqueProcessorOpts

+ func WithHealerTimeout(timeout time.Duration) GoqueProcessorOpts

+ func WithHealerUpdatedAtTimeAgo(updatedAtTimeAgo time.Duration) GoqueProcessorOpts

+ func WithHooksAfterProcessing(hooks ...HookAfterProcessing) GoqueProcessorOpts

+ func WithHooksBeforeProcessing(hooks ...HookBeforeProcessing) GoqueProcessorOpts

+ func WithTaskFetcherMaxTasks(maxTasks int64) GoqueProcessorOpts

+ func WithTaskFetcherTick(tick time.Duration) GoqueProcessorOpts

+ func WithTaskFetcherTimeout(timeout time.Duration) GoqueProcessorOpts

+ func WithTaskProcessingMaxAttempts(maxAttempts int32) GoqueProcessorOpts

+ func WithTaskProcessingNextAttemptAtFunc(nextAttemptAt NextAttemptAtFunc) GoqueProcessorOpts

+ func WithTaskProcessingTimeout(timeout time.Duration) GoqueProcessorOpts

+ func WithWorkersCount(count int) GoqueProcessorOpts

+ func WithWorkersPanicHandler(handler func(any)) GoqueProcessorOpts

+ type HookAfterProcessing func(ctx context.Context, task *entity.Task, err error)

+ type HookBeforeProcessing func(ctx context.Context, task *entity.Task)

+ type NextAttemptAtFunc func(currentAttempt int32) time.Time

+ func RoundStepNextAttemptAtFunc(periods []time.Duration) NextAttemptAtFunc

+ func StaticNextAttemptAtFunc(period time.Duration) NextAttemptAtFunc

+ func StepNextAttemptAtFunc(periods []time.Duration) NextAttemptAtFunc

+ type TaskProcessor interface

+ ProcessTask func(ctx context.Context, task *entity.Task) error

+ func NoopTaskProcessor() TaskProcessor

+ type TaskProcessorFunc func(ctx context.Context, task *entity.Task) error

+ func (f TaskProcessorFunc) ProcessTask(ctx context.Context, task *entity.Task) error

+ type TaskStorage interface

+ GetTasksForProcessing func(ctx context.Context, taskType entity.TaskType, maxTasks int64) ([]*entity.Task, error)

+ UpdateTask func(ctx context.Context, taskID uuid.UUID, task *entity.Task) error