worker

Documentation

Index

• Constants
• type Pool
  • func (p *Pool[T]) Submit(event T) bool
  • func (p *Pool[T]) WorkerLogic() component.ComponentWorker
• type PoolBuilder
  • func NewWorkerPoolBuilder[T any](logger zerolog.Logger, store engine.MessageStore, ...) *PoolBuilder[T]
  • func (b *PoolBuilder[T]) Build() *Pool[T]

Constants

const (
	QueuedItemProcessingLog = "processing queued work item"
	QueuedItemProcessedLog  = "finished processing queued work item"
)

Types

type Pool

type Pool[T any] struct {
	// contains filtered or unexported fields
}

Pool is a worker pool that can be used by a higher-level component to manage a set of workers. The workers are managed by the higher-level component, but the worker pool provides the logic for submitting work to the workers and for processing the work. The worker pool is responsible for storing the work until it is processed by a worker.

func (*Pool[T]) Submit

func (p *Pool[T]) Submit(event T) bool

Submit submits work to the worker pool. The submit logic is responsible for submitting the work to the worker pool.

func (*Pool[T]) WorkerLogic

func (p *Pool[T]) WorkerLogic() component.ComponentWorker

WorkerLogic returns a new worker logic that can be added to a component. The worker logic is responsible for processing the work that is submitted to the worker pool. A pool may have multiple workers, but the worker logic is the same for all the workers. Workers are managed by the higher-level component, through component.AddWorker.

type PoolBuilder

type PoolBuilder[T any] struct {
	// contains filtered or unexported fields
}

PoolBuilder is an auxiliary builder for constructing workers with a common inbound queue, where the workers are managed by a higher-level component.

The message store as well as the processing function are specified by the caller. WorkerPoolBuilder does not add any concurrency handling. It is the callers responsibility to make sure that the number of workers concurrently accessing `processingFunc` is compatible with its implementation.

func NewWorkerPoolBuilder

func NewWorkerPoolBuilder[T any](
	logger zerolog.Logger,
	store engine.MessageStore,
	processingFunc func(input T) error,
) *PoolBuilder[T]

NewWorkerPoolBuilder creates a new PoolBuilder, which is an auxiliary builder for constructing workers with a common inbound queue. Arguments: -`processingFunc`: the function for processing the input tasks. -`store`: temporarily stores inbound events until they are processed. Returns: The function returns a `PoolBuilder` instance.

func (*PoolBuilder[T]) Build

func (b *PoolBuilder[T]) Build() *Pool[T]

Build builds a new worker pool. The worker pool is responsible for storing the work until it is processed by a worker.