taskq

README

TaskQ

Simple and powerful goroutines manager.

---

Installing

go get github.com/antonmashko/taskq

TaskQ when and how to use

Usually you need TaskQ for controlling resources of service (goroutines spawning). Example:

taskq := New(10)
taskq.Start()
taskID, err := taskq.Enqueue(context.Background(), taskq.TaskFunc(func(ctx context.Context) error {
        fmt.Println("hello world")
		return nil
	}))

In this example, we created TaskQ with limiting max active goroutines count to 10. Tasks for executing we're adding to TaskQ with Enqueue. Task will be executed when TaskQ will be able to create new worker (goroutine). Goroutine will dispose and next task will be executed in a new goroutine.

Use limit=0 for not limiting goroutines number.

Graceful shutdown

Shutdown and Close gracefully shuts down the TaskQ without interrupting any active tasks. If TaskQ need to finish all tasks in queue, use context ContextWithWait with Shutdown method.

Documentation

Index

• Variables
• type ConcurrentQueue
  • func NewConcurrentQueue() *ConcurrentQueue
  • func (q *ConcurrentQueue) Dequeue(_ context.Context) (Task, error)
  • func (q *ConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)
  • func (q *ConcurrentQueue) Len(_ context.Context) int
• type LimitedConcurrentQueue
  • func NewLimitedConcurrentQueue(size int) *LimitedConcurrentQueue
  • func (q *LimitedConcurrentQueue) Dequeue(ctx context.Context) (Task, error)
  • func (q *LimitedConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)
  • func (q *LimitedConcurrentQueue) Len(_ context.Context) int
• type 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(limit int) *TaskQ
  • func NewWithQueue(limit int, q Queue) *TaskQ
  • func Pool(size int) *TaskQ
  • func PoolWithQueue(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 (wg *WaitGroup) Enqueue(ctx context.Context, t Task) (int64, error)
  • func (wg *WaitGroup) Wait()

Variables

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

var ContextWithWait = context.WithValue(context.Background(), ctxWaitKey{}, true)

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 NewConcurrentQueue

func NewConcurrentQueue() *ConcurrentQueue

func (*ConcurrentQueue) Dequeue

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

func (*ConcurrentQueue) Enqueue

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

func (*ConcurrentQueue) Len

func (q *ConcurrentQueue) Len(_ context.Context) int

type LimitedConcurrentQueue

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

func NewLimitedConcurrentQueue

func NewLimitedConcurrentQueue(size int) *LimitedConcurrentQueue

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)

func (*LimitedConcurrentQueue) Len

func (q *LimitedConcurrentQueue) Len(_ context.Context) int

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)
}

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 {
	OnDequeueError func(ctx context.Context, workerID uint64, err error)
	// contains filtered or unexported fields
}

func New

func New(limit int) *TaskQ

func NewWithQueue

func NewWithQueue(limit int, q Queue) *TaskQ

func Pool

func Pool(size int) *TaskQ

func PoolWithQueue

func PoolWithQueue(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 {
	// contains filtered or unexported fields
}

func NewWaitGroup

func NewWaitGroup(size int) *WaitGroup

func (*WaitGroup) Enqueue

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

func (*WaitGroup) Wait

func (wg *WaitGroup) Wait()