Method innerFetchLoop

producer.go:637–676

(workCtx context.Context, fetchResultCh chan producerFetchResult)

}

func (p *producer) innerFetchLoop(workCtx context.Context, fetchResultCh chan producerFetchResult) {
	var limit int
	if p.paused {
		limit = 0
	} else {
		limit = p.maxJobsToFetch()
		if limit <= 0 {
			// We have no slots for new jobs, so don't bother fetching. However, since
			// we knew it was time to fetch, we keep track of what happened so we can
			// trigger another fetch as soon as we have open slots.
			p.fetchWhenSlotsAreAvailable = true
			return
		}
	}

	go p.dispatchWork(workCtx, limit, fetchResultCh)

	for {
		select {
		case result := <-fetchResultCh:
			if result.err != nil {
				p.Logger.ErrorContext(workCtx, p.Name+": Error fetching jobs", slog.String("err", result.err.Error()), slog.String("queue", p.config.Queue))
			} else if len(result.jobs) > 0 {
				p.startNewExecutors(workCtx, result.jobs)

				if len(result.jobs) == limit {
					// Fetch returned the maximum number of jobs that were requested,
					// implying there may be more in the queue. Trigger another fetch when
					// slots are available.
					p.fetchWhenSlotsAreAvailable = true
				}
			}
			return
		case result := <-p.jobResultCh:
			p.removeActiveJob(result)
		case jobID := <-p.cancelCh:
			p.maybeCancelJob(workCtx, jobID)
		}
	}
}

func (p *producer) executorShutdownLoop() {
	// No more jobs will be fetched or executed. However, we must wait for all