worker

Documentation

Index

• Variables
• func InputChannelAdapter(nonInterfaceChannel interface{}) (chan interface{}, error)
• type Worker
  • func New(workerFunc WorkerFunc) (*Worker, error)
  • func (w *Worker) GetInputChannel() (chan<- interface{}, error)
  • func (w *Worker) GetOutputChannel() (<-chan interface{}, error)
  • func (w *Worker) SetInputChannel(inputChannel chan interface{}) error
  • func (w *Worker) SetOutputChannel(outputChannel chan interface{}) error
  • func (w *Worker) Start(ctx context.Context) error
• type WorkerError
• type WorkerFunc

Variables

var (
	ErrNotChannel         = errors.New("not a channel")
	ErrNotReadableChannel = errors.New("not a readable channel")
)

var (
	// Errors.
	ErrAlreadyStarted   = errors.New("worker already started")
	ErrNilContext       = errors.New("context must not be nil")
	ErrNilInputChannel  = errors.New("input channel must not be nil")
	ErrNilOutputChannel = errors.New("output channel must not be nil")
	ErrNilWorkerFunc    = errors.New("worker function must not be nil")
	ErrNotStarted       = errors.New("worker not started")
)

Functions

func InputChannelAdapter

func InputChannelAdapter(
	nonInterfaceChannel interface{}) (chan interface{}, error)

InputChannelAdapter expects a readable channel of any value as input and returns a readable channel of interface values. Items are copied from the input channel to the output channel via an internal goroutine.

Types

type Worker

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

Worker implements an asynchronous and interruptible Worker for processing items sent through its input channel and sending results to its output channel.

func New

func New(workerFunc WorkerFunc) (*Worker, error)

New returns a new Worker instance that will use the given workerFunc to process input items.

func (*Worker) GetInputChannel

func (w *Worker) GetInputChannel() (chan<- interface{}, error)

GetInputChannel returns the channel where input items will be read from. The channel will be internally allocated the first time this is called. This must be called at least once before Start() (or SetInputChannel() can be used instead).

This should be used whene there is the need to have multiple Workers processing items from the same source channel to parallelize work.

func (*Worker) GetOutputChannel

func (w *Worker) GetOutputChannel() (<-chan interface{}, error)

GetOutputChannel returns the channel where the result of processing input items will be sent to. The channel will be internally allocated the first time this is called. This must be called at least once before Start() (or SetOutputChannel() can be used instead).

func (*Worker) SetInputChannel

func (w *Worker) SetInputChannel(inputChannel chan interface{}) error

SetInputChannel sets the channel where the Worker will read items from. This must be called before Start() (or GetInputChannel() can be used instead).

func (*Worker) SetOutputChannel

func (w *Worker) SetOutputChannel(outputChannel chan interface{}) error

SetOutputChannel sets the channel where the result of processing input items will be sent to. This must be called at least once before Start() (or GetOutputChannel() can be used instead).

This should be used when there is the need to have multiple Workers sending processed items to the same destination channel, usually to aggregate data that was processed in parallel by multiple Workers.

func (*Worker) Start

func (w *Worker) Start(ctx context.Context) error

Start starts the Worker with the given context. The context can be used to stop Workers with an explicit cancelation or with a timeout and it can also be used to pass required data to the workerFunc.

type WorkerError

type WorkerError struct {
	Item  interface{}
	Error error
}

WorkerError is the error sent through the output channel when there is an error processing an item. It includes the input Item that had an error and the error returned by the WorkerFunc.

type WorkerFunc

type WorkerFunc func(interface{}, context.Context) (interface{}, error)

WorkerFunc is the function type that is used by the Worker to process items. It receives the item to be processed (as an interface{}) and the context the worker was started with (to allow passing extra data to the worker function if needed). WorkerFunc implementations should return the result of processing the input item (also as an in terface{}) and a nil error on success and a nil result and non-nil error on failure.