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() Task
  • func (q *ConcurrentQueue) Enqueue(t Task) int64
• type LimitedConcurrentQueue
  • func (q *LimitedConcurrentQueue) Dequeue() Task
  • func (q *LimitedConcurrentQueue) Enqueue(t Task) int64
• 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(task Task) int64
  • 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(task Task) int64
  • func (wg *WaitGroup) Wait()

Variables

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() Task

func (*ConcurrentQueue) Enqueue

func (q *ConcurrentQueue) Enqueue(t Task) int64

type LimitedConcurrentQueue

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

func (*LimitedConcurrentQueue) Dequeue

func (q *LimitedConcurrentQueue) Dequeue() Task

func (*LimitedConcurrentQueue) Enqueue

func (q *LimitedConcurrentQueue) Enqueue(t Task) int64

type Queue

type Queue interface {
	Enqueue(Task) int64
	Dequeue() Task
}

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(task Task) int64

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(task Task) int64

func (*WaitGroup) Wait

func (wg *WaitGroup) Wait()