workerpool

Documentation

Index

• Variables
• type Job
  • func NewJob(handler JobHandler) *Job
  • func (j *Job) Execute(ctx context.Context) Result
  • func (j *Job) SetStatus(status JobStatus)
  • func (j *Job) Status() JobStatus
• type JobHandler
• type JobReceiver
• type JobSender
• type JobStatus
  • func (j JobStatus) String() string
• type OptionFunc
  • func WithJobCh(jobCh chan *Job) OptionFunc
  • func WithLogger(logger log.Logger) OptionFunc
  • func WithResultCh(resultCh chan Result) OptionFunc
• type Result
• type WorkerPool
  • func New(poolSize int, opts ...OptionFunc) *WorkerPool
  • func (p *WorkerPool) Receive(ctx context.Context) (Result, error)
  • func (p *WorkerPool) Reset()
  • func (p *WorkerPool) Run(ctx context.Context)
  • func (p *WorkerPool) Send(ctx context.Context, jobs ...*Job) error
  • func (p *WorkerPool) Start(ctx context.Context)
  • func (p *WorkerPool) Stop(ctx context.Context)

Variables

var (
	ErrWorkerPoolStopped = errors.New("worker-pool has stopped")

	ErrCannotReadResultChannel = errors.New("cannot read result from a channel")
)

Types

type Job

type Job struct {
	Handler JobHandler
	// contains filtered or unexported fields
}

Job wraps the job-handler that will be executed on a worker. Once the worker consumes the Job from the job-channel, it will execute the job's handler and the obtained execution Result will be published on the result channel

func NewJob

func NewJob(handler JobHandler) *Job

NewJob creates a new Job instance with the job's handler

func (*Job) Execute

func (j *Job) Execute(ctx context.Context) Result

Execute invokes the job's handler

func (*Job) SetStatus

func (j *Job) SetStatus(status JobStatus)

func (*Job) Status

func (j *Job) Status() JobStatus

type JobHandler

type JobHandler func(ctx context.Context) Result

JobHandler is a function that must be wrapped by a Job

type JobReceiver

type JobReceiver interface {
	Receive(ctx context.Context) (Result, error)
}

JobReceiver is an interface that wraps basic job receiver methods

type JobSender

type JobSender interface {
	Send(ctx context.Context, jobs ...Job) error
}

JobSender is an interface that wraps basic job sender methods

type JobStatus

type JobStatus int

const (
	JobCreated JobStatus = iota
	JobSending
	JobSent
	JobReceived
	JobExecuting
	JobExecuted
)

func (JobStatus) String

func (j JobStatus) String() string

String returns string representation of JobStatus

type OptionFunc

type OptionFunc func(*WorkerPool)

OptionFunc is function type for a worker-pool optional functions

func WithJobCh

func WithJobCh(jobCh chan *Job) OptionFunc

WithJobCh sets a job channel to worker-pool using option function

func WithLogger

func WithLogger(logger log.Logger) OptionFunc

WithLogger sets a logger to worker-pool using option function

func WithResultCh

func WithResultCh(resultCh chan Result) OptionFunc

WithResultCh sets a result channel to worker-pool using option function

type Result

type Result struct {
	Value any
	Err   error
}

Result this structure is the result of the job and will be sent to main goroutine via result channel

type WorkerPool

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

WorkerPool is an implementation of a component that allows creating a set of workers to process arbitrary jobs in background

func New

func New(poolSize int, opts ...OptionFunc) *WorkerPool

New creates, initializes and returns a new WorkerPool instance

func (*WorkerPool) Receive

func (p *WorkerPool) Receive(ctx context.Context) (Result, error)

Receive waits for a job result from a worker via result channel

func (*WorkerPool) Reset

func (p *WorkerPool) Reset()

Reset resets some data to initial values, among with these are

func (*WorkerPool) Run

func (p *WorkerPool) Run(ctx context.Context)

Run resets and starts a pool of workers to process jobs

func (*WorkerPool) Send

func (p *WorkerPool) Send(ctx context.Context, jobs ...*Job) error

Send sends one or many jobs to the workers via job channel

func (*WorkerPool) Start

func (p *WorkerPool) Start(ctx context.Context)

Start starts a pool of workers to process jobs

func (*WorkerPool) Stop

func (p *WorkerPool) Stop(ctx context.Context)

Stop stops the worker-pool and all dependent workers