queue

README

go-queue

GoLang in-memory queue implementation based on ants.Pool

Documentation

Index

• type Queue
  • func MakeQueue[ResultType any, TaskArg any](workersCnt int, ...) (*Queue[ResultType, TaskArg], error)
  • func (q *Queue[ResultType, TaskArg]) Enqueue(arg TaskArg) (*Task[ResultType, TaskArg], error)
  • func (q *Queue[ResultType, TaskArg]) EnqueueGroup(args ...TaskArg) (*TaskGroup[ResultType, TaskArg], error)
  • func (q *Queue[ResultType, TaskArg]) GetFreeWorkersCount() int
  • func (q *Queue[ResultType, TaskArg]) GetNumTasksBlocked() int
  • func (q *Queue[ResultType, TaskArg]) GetWorkersCount() int
  • func (q *Queue[ResultType, TaskArg]) SetWorkersCount(workers int)
  • func (q *Queue[ResultType, TaskArg]) Stop()
  • func (q *Queue[ResultType, TaskArg]) StopGraceful(timeout time.Duration) error
• type Task
  • func (t *Task[ResultType, TaskArg]) Done() chan struct{}
  • func (t *Task[ResultType, TaskArg]) GetArg() TaskArg
  • func (t *Task[ResultType, TaskArg]) GetID() int64
  • func (t *Task[ResultType, TaskArg]) Result() chan *TaskResult[ResultType, TaskArg]
• type TaskGroup
  • func (g *TaskGroup[ResultType, TaskArg]) Done() chan struct{}
  • func (g *TaskGroup[ResultType, TaskArg]) GetTasks() []*Task[ResultType, TaskArg]
  • func (g *TaskGroup[ResultType, TaskArg]) Results() chan *TaskResult[ResultType, TaskArg]
• type TaskResult

Types

type Queue

type Queue[ResultType any, TaskArg any] struct {
	// contains filtered or unexported fields
}

func MakeQueue

func MakeQueue[ResultType any, TaskArg any](
	workersCnt int,
	processFunc func(taskId int64, arg TaskArg) (ResultType, error),
	antsOptions ...ants.Option,
) (*Queue[ResultType, TaskArg], error)

func (*Queue[ResultType, TaskArg]) Enqueue

func (q *Queue[ResultType, TaskArg]) Enqueue(arg TaskArg) (*Task[ResultType, TaskArg], error)

Enqueue single task

func (*Queue[ResultType, TaskArg]) EnqueueGroup

func (q *Queue[ResultType, TaskArg]) EnqueueGroup(args ...TaskArg) (*TaskGroup[ResultType, TaskArg], error)

EnqueueGroup of tasks

func (*Queue[ResultType, TaskArg]) GetFreeWorkersCount

func (q *Queue[ResultType, TaskArg]) GetFreeWorkersCount() int

func (*Queue[ResultType, TaskArg]) GetNumTasksBlocked

func (q *Queue[ResultType, TaskArg]) GetNumTasksBlocked() int

func (*Queue[ResultType, TaskArg]) GetWorkersCount

func (q *Queue[ResultType, TaskArg]) GetWorkersCount() int

func (*Queue[ResultType, TaskArg]) SetWorkersCount

func (q *Queue[ResultType, TaskArg]) SetWorkersCount(workers int)

func (*Queue[ResultType, TaskArg]) Stop

func (q *Queue[ResultType, TaskArg]) Stop()

func (*Queue[ResultType, TaskArg]) StopGraceful

func (q *Queue[ResultType, TaskArg]) StopGraceful(timeout time.Duration) error

type Task

type Task[ResultType, TaskArg any] struct {
	// contains filtered or unexported fields
}

func (*Task[ResultType, TaskArg]) Done

func (t *Task[ResultType, TaskArg]) Done() chan struct{}

Done returns channel similar to context.Done()

func (*Task[ResultType, TaskArg]) GetArg

func (t *Task[ResultType, TaskArg]) GetArg() TaskArg

func (*Task[ResultType, TaskArg]) GetID

func (t *Task[ResultType, TaskArg]) GetID() int64

func (*Task[ResultType, TaskArg]) Result

func (t *Task[ResultType, TaskArg]) Result() chan *TaskResult[ResultType, TaskArg]

Result returns the channel from which to read the result

type TaskGroup

type TaskGroup[ResultType, TaskArg any] struct {
	// contains filtered or unexported fields
}

func (*TaskGroup[ResultType, TaskArg]) Done

func (g *TaskGroup[ResultType, TaskArg]) Done() chan struct{}

Done returns channel similar to context.Done()

func (*TaskGroup[ResultType, TaskArg]) GetTasks

func (g *TaskGroup[ResultType, TaskArg]) GetTasks() []*Task[ResultType, TaskArg]

func (*TaskGroup[ResultType, TaskArg]) Results

func (g *TaskGroup[ResultType, TaskArg]) Results() chan *TaskResult[ResultType, TaskArg]

Results returns the channel from which to read the results

type TaskResult

type TaskResult[ResultType, TaskArg any] struct {
	Task   *Task[ResultType, TaskArg]
	Result ResultType
	Error  error
}