Method Start

pkg/queue/queue.go:93–131

Start to enable all worker

()

// Start to enable all worker
func (q *queue) Start() {
	q.routineGroup.Run(func() {
		// Resume tasks in DB
		if len(q.options.resumeTaskType) > 0 && q.taskClient != nil {

			ctx := context.TODO()
			ctx = context.WithValue(ctx, inventory.LoadTaskUser{}, true)
			ctx = context.WithValue(ctx, inventory.LoadUserGroup{}, true)
			tasks, err := q.taskClient.GetPendingTasks(ctx, q.resumeTaskType...)
			if err != nil {
				q.logger.Warning("Failed to get pending tasks from DB for given type %v: %s", q.resumeTaskType, err)
			}

			resumed := 0
			for _, t := range tasks {
				resumedTask, err := NewTaskFromModel(t)
				if err != nil {
					q.logger.Warning("Failed to resume task %d: %s", t.ID, err)
					continue
				}

				if resumedTask.Status() == task.StatusSuspending {
					q.metric.IncSuspendingTask()
					q.metric.IncSubmittedTask()
				}

				if err := q.QueueTask(ctx, resumedTask); err != nil {
					q.logger.Warning("Failed to resume task %d: %s", t.ID, err)
				}
				resumed++
			}

			q.logger.Info("Resumed %d tasks from DB.", resumed)
		}

		q.start()
	})
	q.logger.Info("Queue %q started with %d workers.", q.name, q.workerCount)
}

// Shutdown stops all queues.
func (q *queue) Shutdown() {