Method receive

tools/watch/retrywatcher.go:240–272

receive reads the result from a watcher, restarting it if necessary.

()

// receive reads the result from a watcher, restarting it if necessary.
func (rw *RetryWatcher) receive() {
	defer close(rw.doneChan)
	defer close(rw.resultChan)

	klog.V(4).Info("Starting RetryWatcher.")
	defer klog.V(4).Info("Stopping RetryWatcher.")

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		select {
		case <-rw.stopChan:
			cancel()
			return
		case <-ctx.Done():
			return
		}
	}()

	// We use non sliding until so we don't introduce delays on happy path when WATCH call
	// timeouts or gets closed and we need to reestablish it while also avoiding hot loops.
	wait.NonSlidingUntilWithContext(ctx, func(ctx context.Context) {
		done, retryAfter := rw.doReceive()
		if done {
			cancel()
			return
		}

		time.Sleep(retryAfter)

		klog.V(4).Infof("Restarting RetryWatcher at RV=%q", rw.lastResourceVersion)
	}, rw.minRestartDelay)
}

// ResultChan implements Interface.
func (rw *RetryWatcher) ResultChan() <-chan watch.Event {