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)
• type LimitedConcurrentQueue
  • func (q *LimitedConcurrentQueue) Dequeue() Task
  • func (q *LimitedConcurrentQueue) Enqueue(t Task)
• type Queue
  • func NewConcurrentQueue() Queue
  • func NewLimitedConcurrentQueue(size int) Queue
• type RetryableTask
  • func (t *RetryableTask) Do(ctx context.Context) error
• type Status
  • func TaskStatus(task Task) Status
• type Task
  • func NewRetryableTask(task Task, maxRetries int) Task
• type TaskFunc
  • func (t TaskFunc) Do(ctx context.Context) error
• type TaskManager
  • func NewTaskManger(taskQ *TaskQ) *TaskManager
  • func (m *TaskManager) Close() error
  • func (m *TaskManager) Enqueue(task Task) int64
  • func (m *TaskManager) Shutdown(ctx context.Context) error
  • func (m *TaskManager) Start() error
  • func (m *TaskManager) Task(id int64) Task
• 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 TaskQInterface
• type WaitGroup
  • func ConvertToWaitGroup(taskq TaskQInterface) *WaitGroup
  • func NewWaitGroup(size int) *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)

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)

type Queue

type Queue interface {
	Enqueue(Task)
	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 Status

type Status byte

const (
	None Status = iota
	Pending
	InProgress
	Done
	Failed
)

func TaskStatus

func TaskStatus(task Task) Status

type Task

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

Task for TaskQ

func NewRetryableTask

func NewRetryableTask(task Task, maxRetries int) Task

type TaskFunc

type TaskFunc func(ctx context.Context) error

func (TaskFunc) Do

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

type TaskManager

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

func NewTaskManger

func NewTaskManger(taskQ *TaskQ) *TaskManager

func (*TaskManager) Close

func (m *TaskManager) Close() error

func (*TaskManager) Enqueue

func (m *TaskManager) Enqueue(task Task) int64

func (*TaskManager) Shutdown

func (m *TaskManager) Shutdown(ctx context.Context) error

func (*TaskManager) Start

func (m *TaskManager) Start() error

func (*TaskManager) Task

func (m *TaskManager) Task(id int64) Task

type TaskQ

type TaskQ struct {
	TaskDone   func(int64, Task)
	TaskFailed func(int64, Task, error)
	// 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 TaskQInterface

type TaskQInterface interface {
	Enqueue(Task) int64
	Start() error
	Shutdown(ctx context.Context) error
	Close() error
}

type WaitGroup

type WaitGroup struct {
	TaskQInterface
	// contains filtered or unexported fields
}

func ConvertToWaitGroup

func ConvertToWaitGroup(taskq TaskQInterface) *WaitGroup

func NewWaitGroup

func NewWaitGroup(size int) *WaitGroup

func (*WaitGroup) Enqueue

func (wg *WaitGroup) Enqueue(task Task) int64

func (*WaitGroup) Wait

func (wg *WaitGroup) Wait()