workerpool

Jun 29, 2026

Changes in this version

+ var ErrWorkerPoolResultChannelIsClosed = errors.New("worker job is already closed")

+ func ConsumeResultStream[T any](ctx context.Context, job JobResultPipe[T], consumer func(T) error) error

+ func SafeChannelWrite[T any](ctx context.Context, ch chan<- JobResult[T], value JobResult[T]) error

+ func SubmitJob[T any](ctx context.Context, m Manager, job Job[T]) error

+ type Job interface

+ CanRun func() bool

+ F func() func(ctx context.Context, result JobResultPipe[T]) error

+ ID func() string

+ IncreaseRuns func()

+ Retries func() int

+ Runs func() int

+ func NewJobWithBufferAndRetry[T any](process func(ctx context.Context, result JobResultPipe[T]) error, ...) Job[T]

+ func NewJobWithBuffer[T any](process func(ctx context.Context, result JobResultPipe[T]) error, buffer int) Job[T]

+ func NewJobWithRetry[T any](process func(ctx context.Context, result JobResultPipe[T]) error, retries int) Job[T]

+ func NewJob[T any](process func(ctx context.Context, result JobResultPipe[T]) error) Job[T]

+ type JobImpl struct

+ func (ji *JobImpl[T]) CanRun() bool

+ func (ji *JobImpl[T]) Close()

+ func (ji *JobImpl[T]) F() func(ctx context.Context, result JobResultPipe[T]) error

+ func (ji *JobImpl[T]) ID() string

+ func (ji *JobImpl[T]) IncreaseRuns()

+ func (ji *JobImpl[T]) ReadResult(ctx context.Context) (JobResult[T], bool)

+ func (ji *JobImpl[T]) ResultBufferSize() int

+ func (ji *JobImpl[T]) ResultChan() <-chan JobResult[T]

+ func (ji *JobImpl[T]) Retries() int

+ func (ji *JobImpl[T]) Runs() int

+ func (ji *JobImpl[T]) WriteError(ctx context.Context, val error) error

+ func (ji *JobImpl[T]) WriteResult(ctx context.Context, val T) error

+ type JobResult interface

+ Error func() error

+ IsError func() bool

+ Item func() T

+ func ErrorResult[T any](err error) JobResult[T]

+ func Result[T any](item T) JobResult[T]

+ func SafeChannelRead[T any](ctx context.Context, ch <-chan JobResult[T]) (JobResult[T], bool)

+ type JobResultPipe interface

+ Close func()

+ ReadResult func(ctx context.Context) (JobResult[T], bool)

+ ResultBufferSize func() int

+ ResultChan func() <-chan JobResult[T]

+ WriteError func(ctx context.Context, val error) error

+ WriteResult func(ctx context.Context, val T) error

+ type Manager interface

+ GetPool func() (WorkerPool, error)

+ Shutdown func(context.Context) error

+ StopError func(context.Context, error)

+ func NewManager(ctx context.Context, cfg config.ConfigurationWorkerPool, ...) (Manager, error)

+ type Option func(*Options)

+ func WithConcurrency(concurrency int) Option

+ func WithPoolCount(count int) Option

+ func WithPoolDisablePurge(disable bool) Option

+ func WithPoolExpiryDuration(duration time.Duration) Option

+ func WithPoolLogger(logger *util.LogEntry) Option

+ func WithPoolNonblocking(nonblocking bool) Option

+ func WithPoolPanicHandler(handler func(any)) Option

+ func WithPoolPreAlloc(preAlloc bool) Option

+ func WithSinglePoolCapacity(capacity int) Option

+ type Options struct

+ Concurrency int

+ DisablePurge bool

+ ExpiryDuration time.Duration

+ Logger *util.LogEntry

+ Nonblocking bool

+ PanicHandler func(any)

+ PoolCount int

+ PreAlloc bool

+ SinglePoolCapacity int

+ type WorkerPool interface

+ Shutdown func()

+ Submit func(ctx context.Context, task func()) error