Method work

pkg/queue/queue.go:221–266

(t Task)

}

func (q *queue) work(t Task) {
	ctx := q.newContext(t)
	l := logging.FromContext(ctx)
	timeIterationStart := time.Now()

	var err error
	// to handle panic cases from inside the worker
	// in such case, we start a new goroutine
	defer func() {
		q.metric.DecBusyWorker()
		e := recover()
		if e != nil {
			l.Error("Panic error in queue %q: %v", q.name, e)
			t.OnError(fmt.Errorf("panic error: %v", e), time.Since(timeIterationStart))

			_ = q.transitStatus(ctx, t, task.StatusError)
		}
		q.schedule()
	}()

	err = q.transitStatus(ctx, t, task.StatusProcessing)
	if err != nil {
		l.Error("failed to transit task %d to processing: %s", t.ID(), err.Error())
		panic(err)
	}

	for {
		timeIterationStart = time.Now()
		var next task.Status
		next, err = q.run(ctx, t)
		if err != nil {
			t.OnError(err, time.Since(timeIterationStart))
			l.Error("runtime error in queue %q: %s", q.name, err.Error())

			_ = q.transitStatus(ctx, t, task.StatusError)
			break
		}

		// iteration completes
		t.OnIterationComplete(time.Since(timeIterationStart))
		_ = q.transitStatus(ctx, t, next)
		if next != task.StatusProcessing {
			break
		}
	}
}

func (q *queue) run(ctx context.Context, t Task) (task.Status, error) {
	l := logging.FromContext(ctx)