Method dispatchWork

producer.go:790–819

(workCtx context.Context, count int, fetchResultCh chan<- producerFetchResult)

}

func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultCh chan<- producerFetchResult) {
	// This intentionally removes any deadlines or cancellation from the parent
	// context because we don't want it to get cancelled if the producer is asked
	// to shut down. In that situation, we want to finish fetching any jobs we are
	// in the midst of fetching, work them, and then stop. Otherwise we'd have a
	// risk of shutting down when we had already fetched jobs in the database,
	// leaving those jobs stranded. We'd then potentially have to release them
	// back to the queue.
	ctx := context.WithoutCancel(workCtx)

	// Maximum size of the `attempted_by` array on each job row. This maximum is
	// rarely hit, but exists to protect against degenerate cases.
	const maxAttemptedBy = 100

	jobs, err := p.pilot.JobGetAvailable(ctx, p.exec, p.state, &riverdriver.JobGetAvailableParams{
		ClientID:       p.config.ClientID,
		MaxAttemptedBy: maxAttemptedBy,
		MaxToLock:      count,
		Now:            p.Time.NowOrNil(),
		Queue:          p.config.Queue,
		ProducerID:     p.id.Load(),
		Schema:         p.config.Schema,
	})
	if err != nil {
		fetchResultCh <- producerFetchResult{err: err}
		return
	}

	fetchResultCh <- producerFetchResult{jobs: jobs}
}

// Periodically logs an informational log line giving some insight into the
// current state of the producer.