taskqueue

Documentation

Index

• type Interface
  • func New(id string, size int, workFunc WorkerFunc, failFast bool, wg *sync.WaitGroup) (Interface, error)
• type QueueController
• type QueueControllerCollection
  • func NewQueueControllerCollection(waitGroup *sync.WaitGroup, queues ...QueueController) *QueueControllerCollection
  • func (q *QueueControllerCollection) Add(queue QueueController)
  • func (q *QueueControllerCollection) GetErrorList() *multierror.Error
  • func (q *QueueControllerCollection) LogTaskProcessed()
  • func (q *QueueControllerCollection) Start(ctx context.Context)
  • func (q *QueueControllerCollection) Stop()
  • func (q *QueueControllerCollection) Wait()
• type WorkerFunc

Types

type Interface

type Interface interface {
	QueueController
	// AddTask adds a task to the tasks queue and increase wg counter
	// returns true if the task is added and false if it is skipped
	// (e.g. if the TaskQueue is stopped or failFast situation)
	AddTask(task interface{}) bool
}

Interface can Start/Stop the queue and see its status

func New

func New(id string, size int, workFunc WorkerFunc, failFast bool, wg *sync.WaitGroup) (Interface, error)

New create an empty task queue

type QueueController

type QueueController interface {
	// Start initializes worker's goroutines. The provided context ctx is used by worker goroutines
	Start(ctx context.Context)
	// Stop stops the worker's goroutines, it could be triggered internally on context cancellation or failFast situation
	Stop()
	// GetErrorList returns the errors, occurred during task processing
	GetErrorList() *multierror.Error
	// Name used when logging information
	Name() string
	// GetProcessedTasksCount returns the processed tasks count
	GetProcessedTasksCount() int
	// GetWaitingTasksCount returns waiting tasks count
	GetWaitingTasksCount() int
}

QueueController can Start/Stop the queue and see its status

type QueueControllerCollection

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

QueueControllerCollection holds multiple queue controllers

func NewQueueControllerCollection

func NewQueueControllerCollection(waitGroup *sync.WaitGroup, queues ...QueueController) *QueueControllerCollection

NewQueueControllerCollection creates a new QueueControllerCollection

func (*QueueControllerCollection) Add

func (q *QueueControllerCollection) Add(queue QueueController)

Add adds a controller

func (*QueueControllerCollection) GetErrorList

func (q *QueueControllerCollection) GetErrorList() *multierror.Error

GetErrorList returns error list

func (*QueueControllerCollection) LogTaskProcessed

func (q *QueueControllerCollection) LogTaskProcessed()

LogTaskProcessed logs task processed

func (*QueueControllerCollection) Start

func (q *QueueControllerCollection) Start(ctx context.Context)

Start starts all queues

func (*QueueControllerCollection) Stop

func (q *QueueControllerCollection) Stop()

Stop stops all queues

func (*QueueControllerCollection) Wait

func (q *QueueControllerCollection) Wait()

Wait waits for all queues to finish

type WorkerFunc

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

The WorkerFunc type declares workers functional interface