taskq

README

TaskQ

Goroutine manager.

---

Installing

go get github.com/antonmashko/taskq

TaskQ

Initializing

Use TaskQ for managing you goroutines.

taskq := New(<size>)

size - parameter will control goroutines count. In case if all goroutines are busy, task will be added to queue and will wait for a free worker (goroutine) from pool.

Add task to TaskQ

taskID := taskq.Enqueue(<task>)

Start TaskQ

err := taskq.Start()

run all added and future tasks in taskq.

Documentation

Index

• Variables
• type ConcurrentQueue
  • func (q *ConcurrentQueue) Dequeue(_ context.Context) (Task, error)
  • func (q *ConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)
• type LimitedConcurrentQueue
  • func (q *LimitedConcurrentQueue) Dequeue(ctx context.Context) (Task, error)
  • func (q *LimitedConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)
• type Queue
  • func NewConcurrentQueue() Queue
  • func NewLimitedConcurrentQueue(size int) Queue
• type RetryableTask
  • func (t *RetryableTask) Do(ctx context.Context) error
• type Task
  • func NewRetryableTask(task Task, maxRetries int) Task
• type TaskDone
• type TaskFunc
  • func (t TaskFunc) Do(ctx context.Context) error
• type TaskOnError
• type TaskQ
  • func New(size int) *TaskQ
  • func NewWithQueue(size int, q Queue) *TaskQ
  • func (t *TaskQ) Close() error
  • func (t *TaskQ) Enqueue(ctx context.Context, task Task) (int64, error)
  • func (t *TaskQ) Shutdown(ctx context.Context) error
  • func (t *TaskQ) Start() error
• type WaitGroup
  • func NewWaitGroup(size int) *WaitGroup
  • func NewWaitGroupFromTaskq(taskq *TaskQ) *WaitGroup
  • func (wg *WaitGroup) Enqueue(ctx context.Context, task Task) (int64, error)
  • func (wg *WaitGroup) Wait()

Variables

var (
	ErrClosed  = errors.New("taskq closed")
	ErrNilTask = errors.New("nil task")
)

var (
	EmptyQueue = errors.New("empty queue")
)

var ErrRetryTask = errors.New("error occurred during task execution. retry task")

Types

type ConcurrentQueue

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

func (*ConcurrentQueue) Dequeue

func (q *ConcurrentQueue) Dequeue(_ context.Context) (Task, error)

func (*ConcurrentQueue) Enqueue

func (q *ConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)

type LimitedConcurrentQueue

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

func (*LimitedConcurrentQueue) Dequeue

func (q *LimitedConcurrentQueue) Dequeue(ctx context.Context) (Task, error)

func (*LimitedConcurrentQueue) Enqueue

func (q *LimitedConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)

type Queue

type Queue interface {
	Enqueue(context.Context, Task) (int64, error)
	// Dequeue Task from queue
	// if queue empty return `EmptyQueue` as error
	Dequeue(context.Context) (Task, error)
}

func NewConcurrentQueue

func NewConcurrentQueue() Queue

func NewLimitedConcurrentQueue

func NewLimitedConcurrentQueue(size int) Queue

type RetryableTask

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

func (*RetryableTask) Do

func (t *RetryableTask) Do(ctx context.Context) error

type Task

type Task interface {
	Do(ctx context.Context) error
}

Task for TaskQ

func NewRetryableTask

func NewRetryableTask(task Task, maxRetries int) Task

type TaskDone

type TaskDone interface {
	Done(context.Context)
}

type TaskFunc

type TaskFunc func(ctx context.Context) error

func (TaskFunc) Do

func (t TaskFunc) Do(ctx context.Context) error

type TaskOnError

type TaskOnError interface {
	OnError(context.Context, error)
}

type TaskQ

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

func New

func New(size int) *TaskQ

func NewWithQueue

func NewWithQueue(size int, q Queue) *TaskQ

func (*TaskQ) Close

func (t *TaskQ) Close() error

func (*TaskQ) Enqueue

func (t *TaskQ) Enqueue(ctx context.Context, task Task) (int64, error)

func (*TaskQ) Shutdown

func (t *TaskQ) Shutdown(ctx context.Context) error

func (*TaskQ) Start

func (t *TaskQ) Start() error

type WaitGroup

type WaitGroup struct {
	*TaskQ
	// contains filtered or unexported fields
}

func NewWaitGroup

func NewWaitGroup(size int) *WaitGroup

func NewWaitGroupFromTaskq

func NewWaitGroupFromTaskq(taskq *TaskQ) *WaitGroup

func (*WaitGroup) Enqueue

func (wg *WaitGroup) Enqueue(ctx context.Context, task Task) (int64, error)

func (*WaitGroup) Wait

func (wg *WaitGroup) Wait()