Method Start

internal/notifier/notifier.go:109–166

(ctx context.Context)

}

func (n *Notifier) Start(ctx context.Context) error {
	ctx, shouldStart, started, stopped := n.StartInit(ctx)
	if !shouldStart {
		return nil
	}

	// The loop below will connect/close on every iteration, but do one initial
	// connect so the notifier fails fast in case of an obvious problem.
	if err := n.listenerConnect(ctx, false); err != nil {
		stopped()
		if errors.Is(err, context.Canceled) {
			return nil
		}
		return err
	}

	go func() {
		started()
		defer stopped()

		n.Logger.DebugContext(ctx, n.Name+": Run loop started")
		defer n.Logger.DebugContext(ctx, n.Name+": Run loop stopped")

		n.withLock(func() { n.isStarted = true })
		defer n.withLock(func() { n.isStarted = false })

		defer n.listenerClose(ctx, false)

		var wg sync.WaitGroup

		wg.Go(func() {
			n.deliverNotifications(ctx)
		})

		for attempt := 0; ; attempt++ {
			if err := n.listenAndWait(ctx); err != nil {
				if errors.Is(err, context.Canceled) {
					break
				}

				sleepDuration := serviceutil.ExponentialBackoff(attempt, serviceutil.MaxAttemptsBeforeResetDefault)
				n.Logger.ErrorContext(ctx, n.Name+": Error running listener (will attempt reconnect after backoff)",
					slog.Int("attempt", attempt),
					slog.String("err", err.Error()),
					slog.String("sleep_duration", sleepDuration.String()),
				)
				n.testSignals.BackoffError.Signal(err)
				if !n.testDisableSleep {
					serviceutil.CancellableSleep(ctx, sleepDuration)
				}
			}
		}

		wg.Wait()
	}()

	return nil
}