queue

Documentation

Index

• Constants
• Variables
• type TaskQueue
  • func NewTasksQueue() *TaskQueue
  • func (q *TaskQueue) AddAfter(id string, newTask task.Task)
  • func (q *TaskQueue) AddBefore(id string, newTask task.Task)
  • func (q *TaskQueue) AddFirst(t task.Task)
  • func (q *TaskQueue) AddLast(t task.Task)
  • func (q *TaskQueue) CancelTaskDelay()
  • func (q *TaskQueue) Filter(filterFn func(task.Task) bool)
  • func (q *TaskQueue) Get(id string) (t task.Task)
  • func (q *TaskQueue) GetFirst() task.Task
  • func (q *TaskQueue) GetLast() (t task.Task)
  • func (q *TaskQueue) IsEmpty() bool
  • func (q *TaskQueue) Iterate(doFn func(task.Task))
  • func (q *TaskQueue) Length() int
  • func (q *TaskQueue) MeasureActionTime(action string) func()
  • func (q *TaskQueue) Remove(id string) (t task.Task)
  • func (q *TaskQueue) RemoveFirst() (t task.Task)
  • func (q *TaskQueue) RemoveLast() (t task.Task)
  • func (q *TaskQueue) SetDebug(debug bool)
  • func (q *TaskQueue) Start()
  • func (q *TaskQueue) Stop()
  • func (q *TaskQueue) String() string
  • func (q *TaskQueue) WithContext(ctx context.Context)
  • func (tq *TaskQueue) WithHandler(fn func(task.Task) TaskResult) *TaskQueue
  • func (q *TaskQueue) WithMetricStorage(mstor *metric_storage.MetricStorage)
  • func (tq *TaskQueue) WithName(name string) *TaskQueue
• type TaskQueueSet
  • func NewTaskQueueSet() *TaskQueueSet
  • func (tqs *TaskQueueSet) Add(queue *TaskQueue)
  • func (tqs *TaskQueueSet) DoWithLock(fn func(tqs *TaskQueueSet))
  • func (tqs *TaskQueueSet) GetByName(name string) *TaskQueue
  • func (tqs *TaskQueueSet) GetMain() *TaskQueue
  • func (tqs *TaskQueueSet) Iterate(doFn func(queue *TaskQueue))
  • func (tqs *TaskQueueSet) NewNamedQueue(name string, handler func(task.Task) TaskResult)
  • func (tqs *TaskQueueSet) Remove(name string)
  • func (tqs *TaskQueueSet) Start()
  • func (tqs *TaskQueueSet) StartMain()
  • func (tqs *TaskQueueSet) Stop()
  • func (tqs *TaskQueueSet) WaitStopWithTimeout(timeout time.Duration)
  • func (tqs *TaskQueueSet) WithContext(ctx context.Context)
  • func (tqs *TaskQueueSet) WithMainName(name string)
  • func (tqs *TaskQueueSet) WithMetricStorage(mstor *metric_storage.MetricStorage)
• type TaskResult
• type TaskStatus

Constants

const MainQueueName = "main"

Variables

var (
	DefaultWaitLoopCheckInterval    = 125 * time.Millisecond
	DefaultDelayOnQueueIsEmpty      = 250 * time.Millisecond
	DefaultInitialDelayOnFailedTask = 5 * time.Second
	DefaultDelayOnRepeat            = 25 * time.Millisecond
)

Types

type TaskQueue

type TaskQueue struct {
	Name    string
	Handler func(task.Task) TaskResult
	Status  string

	// Timing settings.
	WaitLoopCheckInterval time.Duration
	DelayOnQueueIsEmpty   time.Duration
	DelayOnRepeat         time.Duration
	ExponentialBackoffFn  func(failureCount int) time.Duration
	// contains filtered or unexported fields
}

func NewTasksQueue

func NewTasksQueue() *TaskQueue

func (*TaskQueue) AddAfter

func (q *TaskQueue) AddAfter(id string, newTask task.Task)

AddAfter inserts a task after the task with specified id.

func (*TaskQueue) AddBefore

func (q *TaskQueue) AddBefore(id string, newTask task.Task)

AddBefore inserts a task before the task with specified id.

func (*TaskQueue) AddFirst

func (q *TaskQueue) AddFirst(t task.Task)

AddFirst adds new head element.

func (*TaskQueue) AddLast

func (q *TaskQueue) AddLast(t task.Task)

AddFirst adds new tail element.

func (*TaskQueue) CancelTaskDelay

func (q *TaskQueue) CancelTaskDelay()

CancelTaskDelay breaks wait loop. Useful to break the possible long sleep delay.

func (*TaskQueue) Filter

func (q *TaskQueue) Filter(filterFn func(task.Task) bool)

Filter run filterFn on every task and remove each with false result.

func (*TaskQueue) Get

func (q *TaskQueue) Get(id string) (t task.Task)

Get returns a task by id.

func (*TaskQueue) GetFirst

func (q *TaskQueue) GetFirst() task.Task

GetFirst returns a head element.

func (*TaskQueue) GetLast

func (q *TaskQueue) GetLast() (t task.Task)

GetLast returns a tail element.

func (*TaskQueue) IsEmpty

func (q *TaskQueue) IsEmpty() bool

func (*TaskQueue) Iterate

func (q *TaskQueue) Iterate(doFn func(task.Task))

Iterate run doFn for every task.

func (*TaskQueue) Length

func (q *TaskQueue) Length() int

func (*TaskQueue) MeasureActionTime

func (q *TaskQueue) MeasureActionTime(action string) func()

MeasureActionTime is a helper to measure execution time of queue's actions

func (*TaskQueue) Remove

func (q *TaskQueue) Remove(id string) (t task.Task)

Remove finds element by id and deletes it.

func (*TaskQueue) RemoveFirst

func (q *TaskQueue) RemoveFirst() (t task.Task)

RemoveFirst deletes a head element, so head is moved.

func (*TaskQueue) RemoveLast

func (q *TaskQueue) RemoveLast() (t task.Task)

RemoveLast deletes a tail element, so tail is moved.

func (*TaskQueue) SetDebug

func (q *TaskQueue) SetDebug(debug bool)

func (*TaskQueue) Start

func (q *TaskQueue) Start()

func (*TaskQueue) Stop

func (q *TaskQueue) Stop()

func (*TaskQueue) String

func (q *TaskQueue) String() string

Dump tasks in queue to one line

func (*TaskQueue) WithContext

func (q *TaskQueue) WithContext(ctx context.Context)

func (*TaskQueue) WithHandler

func (tq *TaskQueue) WithHandler(fn func(task.Task) TaskResult) *TaskQueue

func (*TaskQueue) WithMetricStorage

func (q *TaskQueue) WithMetricStorage(mstor *metric_storage.MetricStorage)

func (*TaskQueue) WithName

func (tq *TaskQueue) WithName(name string) *TaskQueue

type TaskQueueSet

type TaskQueueSet struct {
	Queues   map[string]*TaskQueue
	MainName string
	// contains filtered or unexported fields
}

TaskQueueSet is a manager for a set of named queues

func NewTaskQueueSet

func NewTaskQueueSet() *TaskQueueSet

func (*TaskQueueSet) Add

func (tqs *TaskQueueSet) Add(queue *TaskQueue)

func (*TaskQueueSet) DoWithLock

func (tqs *TaskQueueSet) DoWithLock(fn func(tqs *TaskQueueSet))

*

taskQueueSet.DoWithLock(func(tqs *TaskQueueSet){
   tqs.GetMain().Pop()
})

func (*TaskQueueSet) GetByName

func (tqs *TaskQueueSet) GetByName(name string) *TaskQueue

func (*TaskQueueSet) GetMain

func (tqs *TaskQueueSet) GetMain() *TaskQueue

func (*TaskQueueSet) Iterate

func (tqs *TaskQueueSet) Iterate(doFn func(queue *TaskQueue))

Iterate run doFn for every task.

func (*TaskQueueSet) NewNamedQueue

func (tqs *TaskQueueSet) NewNamedQueue(name string, handler func(task.Task) TaskResult)

func (*TaskQueueSet) Remove

func (tqs *TaskQueueSet) Remove(name string)

func (*TaskQueueSet) Start

func (tqs *TaskQueueSet) Start()

func (*TaskQueueSet) StartMain

func (tqs *TaskQueueSet) StartMain()

func (*TaskQueueSet) Stop

func (tqs *TaskQueueSet) Stop()

func (*TaskQueueSet) WaitStopWithTimeout

func (tqs *TaskQueueSet) WaitStopWithTimeout(timeout time.Duration)

func (*TaskQueueSet) WithContext

func (tqs *TaskQueueSet) WithContext(ctx context.Context)

func (*TaskQueueSet) WithMainName

func (tqs *TaskQueueSet) WithMainName(name string)

func (*TaskQueueSet) WithMetricStorage

func (tqs *TaskQueueSet) WithMetricStorage(mstor *metric_storage.MetricStorage)

type TaskResult

type TaskResult struct {
	Status     TaskStatus
	HeadTasks  []task.Task
	TailTasks  []task.Task
	AfterTasks []task.Task

	DelayBeforeNextTask time.Duration

	AfterHandle func()
}

type TaskStatus

type TaskStatus string

const (
	Success TaskStatus = "Success"
	Fail    TaskStatus = "Fail"
	Repeat  TaskStatus = "Repeat"
	Keep    TaskStatus = "Keep"
)