Method Execute

internal/jobexecutor/job_executor.go:147–178

(ctx context.Context)

}

func (e *JobExecutor) Execute(ctx context.Context) {
	// Ensure that the context is cancelled no matter what, or it will leak:
	defer e.CancelFunc(errExecutorDefaultCancel)

	e.start = e.Time.Now()
	e.stats = &jobstats.JobStatistics{
		QueueWaitDuration: e.start.Sub(e.JobRow.ScheduledAt),
	}

	res := e.execute(ctx)
	if res.Err != nil && errors.Is(context.Cause(ctx), rivertype.ErrJobCancelledRemotely) {
		res.Err = context.Cause(ctx)
	}

	var multiJobErrors withJobsAndErrorsByID
	if res.Err != nil {
		multiJobErrors, _ = res.Err.(withJobsAndErrorsByID)
	}

	if multiJobErrors == nil {
		e.reportResult(ctx, e.JobRow, res)
	} else {
		errorsByID := multiJobErrors.ErrorsByID()
		for _, jobRow := range multiJobErrors.Jobs() {
			jobSpecificRes := *res
			jobSpecificRes.Err = errorsByID[jobRow.ID]
			e.reportResult(ctx, jobRow, &jobSpecificRes)
		}
	}

	e.ProducerCallbacks.JobDone(e.JobRow)
}

// Executes the job, handling a panic if necessary (and various other error
// conditions). The named return value is so that we can still return a value in