Method pollForSettingChanges

producer.go:912–991

(ctx context.Context, wg *sync.WaitGroup, lastPaused bool, lastMetadata []byte)

}

func (p *producer) pollForSettingChanges(ctx context.Context, wg *sync.WaitGroup, lastPaused bool, lastMetadata []byte) {
	defer wg.Done()

	ticker := time.NewTicker(p.config.QueuePollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			updatedQueue, err := func() (*rivertype.Queue, error) {
				ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
				defer cancel()

				return p.exec.QueueGet(ctx, &riverdriver.QueueGetParams{
					Name:   p.config.Queue,
					Schema: p.config.Schema,
				})
			}()
			if err != nil {
				// Don't log if this is part of a standard shutdown.
				if !errors.Is(context.Cause(ctx), startstop.ErrStop) {
					p.Logger.ErrorContext(ctx, p.Name+": Error fetching queue settings", slog.String("err", err.Error()))
				}
				continue
			}

			if updatedQueue == nil {
				p.Logger.ErrorContext(ctx, p.Name+": Queue row not found when polling for setting changes", slog.String("queue", p.config.Queue))
				continue
			}

			// Look for a change in the paused state:
			shouldBePaused := (updatedQueue.PausedAt != nil)
			if lastPaused != shouldBePaused {
				action := controlActionPause
				if !shouldBePaused {
					action = controlActionResume
				}
				payload := &controlEventPayload{
					Action: action,
					Queue:  p.config.Queue,
				}
				p.Logger.DebugContext(ctx, p.Name+": Queue control state changed from polling",
					slog.String("queue", p.config.Queue),
					slog.String("action", string(action)),
					slog.Bool("paused", shouldBePaused),
				)

				select {
				case p.queueControlCh <- payload:
					lastPaused = shouldBePaused
				default:
					p.Logger.WarnContext(ctx, p.Name+": Queue control notification dropped due to full buffer", slog.String("action", string(action)))
				}
			}

			// Look for a change in the queue's metadata: