worker

Documentation

Overview

Package worker provides a generic worker pool for processing queued jobs. This package offers concurrent job processing with configurable worker counts per queue.

Index

• type Config
  • func DefaultConfig() Config
• type JobProcessor
• type Pool
  • func NewPool(queue Queue, processor JobProcessor, config Config) *Pool
  • func (p *Pool) Start()
  • func (p *Pool) Stop()
• type Queue
• type Worker
  • func (w *Worker) Start()

Types

type Config

type Config struct {
	Queues map[string]int // Queue name -> number of workers
}

Config configures the worker pool

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns the default worker configuration

type JobProcessor

type JobProcessor interface {
	Process(ctx context.Context, job interface{}) error
	GetJobID(job interface{}) string
	GetTimeout(job interface{}) time.Duration
}

JobProcessor defines the interface for processing jobs

type Pool

type Pool struct {
	// contains filtered or unexported fields
}

Pool manages a pool of workers that process jobs from queues

func NewPool

func NewPool(queue Queue, processor JobProcessor, config Config) *Pool

NewPool creates a new worker pool

func (*Pool) Start

func (p *Pool) Start()

Start starts all workers in the pool

func (*Pool) Stop

func (p *Pool) Stop()

Stop stops all workers in the pool

type Queue

type Queue interface {
	Dequeue(queueName string, timeout time.Duration) (interface{}, error)
	Enqueue(job interface{}) error
	MarkProcessing(jobID string, deadline time.Time) error
	CompleteJob(jobID string) error
	FailJob(jobID string, requeue bool, queueName string, retryCount int) error
}

Queue defines the interface for job queue operations

type Worker

type Worker struct {
	// contains filtered or unexported fields
}

Worker represents a single worker that processes jobs from a queue

func (*Worker) Start

func (w *Worker) Start()

Start starts a worker processing loop