Function NewTasksQueue

pkg/task/queue/task_queue.go:180–208

NewTasksQueue creates a new TaskQueue with the provided options

(name string, metricStorage metricsstorage.Storage, opts ...TaskQueueOption)

// NewTasksQueue creates a new TaskQueue with the provided options
func NewTasksQueue(name string, metricStorage metricsstorage.Storage, opts ...TaskQueueOption) *TaskQueue {
	q := &TaskQueue{
		Name:    name,
		storage: newTaskStorage(),
		// Default timings
		WaitLoopCheckInterval: DefaultWaitLoopCheckInterval,
		DelayOnQueueIsEmpty:   DefaultDelayOnQueueIsEmpty,
		DelayOnRepeat:         DefaultDelayOnRepeat,
		ExponentialBackoffFn: func(failureCount int) time.Duration {
			return exponential_backoff.CalculateDelay(DefaultInitialDelayOnFailedTask, failureCount)
		},
		logger: log.NewLogger().Named("task_queue").Named(name),
		// Pre-allocate buffers
		contextBuffer:   make([]bindingcontext.BindingContext, 0, 128),
		monitorIDBuffer: make([]string, 0, 128),
		groupBuffer:     make(map[string]*compactionGroup, 32),
		metricStorage:   metricStorage,
		status:          NewTaskQueueStatus(),
	}

	// Apply all options
	for _, opt := range opts {
		opt(q)
	}

	q.queueTasksCounter = NewTaskCounter(q.Name, q.compactableTypes, q.metricStorage)

	return q
}

func (q *TaskQueue) getCompactionGroup() *compactionGroup {
	return compactionGroupPool.Get().(*compactionGroup)