processor

package

Go to main page

Versions in this module

v0

v0.0.2

Nov 28, 2025

Changes in this version

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

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

type GoqueProcessor

+ func (p *GoqueProcessor) Tune(opts []GoqueProcessorOpts)

type GoqueProcessorOpts

+ func GetGoqueProcessorOpts(opts []commonopts.InternalProcessorOpt) []GoqueProcessorOpts

+ func WithHooksAfterProcessing(hooks ...HookAfterProcessing) GoqueProcessorOpts

+ func WithHooksBeforeProcessing(hooks ...HookBeforeProcessing) GoqueProcessorOpts

+ func WithReplaceHooksAfterProcessing(hooks ...HookAfterProcessing) GoqueProcessorOpts

+ func WithReplaceHooksBeforeProcessing(hooks ...HookBeforeProcessing) GoqueProcessorOpts

+ func WithTaskFetcher(fetcher TaskFetcher) GoqueProcessorOpts

+ func WithTaskFetcherMaxTasks(maxTasks int64) GoqueProcessorOpts

+ func WithTaskFetcherTick(tick time.Duration) GoqueProcessorOpts

+ func WithTaskFetcherTimeout(timeout time.Duration) GoqueProcessorOpts

+ func (o GoqueProcessorOpts) OptionType() string

+ type TaskFetcher interface

+ FetchTasks func(ctx context.Context) ([]*entity.Task, error)

+ type TaskFetcherFunc func(ctx context.Context) ([]*entity.Task, error)

+ func (f TaskFetcherFunc) FetchTasks(ctx context.Context) ([]*entity.Task, error)

type TaskProcessor

+ func NoopTaskProcessor() TaskProcessor

v0.0.1

Nov 28, 2025

Changes in this version

+ var ErrTaskCancel = fmt.Errorf("task canceled")

+ var ErrTaskTimeout = fmt.Errorf("task processing timeout")

+ var FibonacciPeriods = []time.Duration

+ var StrongPeriods = []time.Duration

+ type GoqueProcessor struct

+ func NewGoqueProcessor(taskRepo TaskStorage, taskType entity.TaskType, taskProcessor TaskProcessor, ...) *GoqueProcessor

+ func (p *GoqueProcessor) Name() string

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

+ func (p *GoqueProcessor) Stop()

+ type GoqueProcessorOpts func(*GoqueProcessor)

+ func WithFetcherMaxTasks(maxTasks int64) GoqueProcessorOpts

+ func WithFetcherTick(tick time.Duration) GoqueProcessorOpts

+ func WithHookAfterProcessing(hooks ...HookAfterProcessing) GoqueProcessorOpts

+ func WithHookBeforeProcessing(hooks ...HookBeforeProcessing) GoqueProcessorOpts

+ func WithMaxAttempts(maxAttempts int32) GoqueProcessorOpts

+ func WithNextAttemptAtFunc(nextAttemptAt NextAttemptAtFunc) GoqueProcessorOpts

+ func WithStaticNextAttemptAtFunc(period time.Duration) GoqueProcessorOpts

+ func WithTaskTimeout(taskTimeout time.Duration) GoqueProcessorOpts

+ func WithWorkersCount(count int) 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, payload string) error

+ type TaskProcessorFunc func(ctx context.Context, payload string) error

+ func (f TaskProcessorFunc) ProcessTask(ctx context.Context, payload string) error

+ type TaskStorage interface

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

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

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