// must always be called, even if the consumer has not yet called
	// ResultChan().
	//
	// Only the consumer should call Stop(), not the producer. If the producer
	// errors and needs to stop the watch prematurely, it should instead send
	// an error event and close the result channel.
	Stop()

	// ResultChan returns a channel which will receive events from the event

	case e, ok := <-w.ResultChan():
		if ok {
			var obj string
			switch e.Object.(type) {
			case *example.Pod:
				obj = e.Object.(*example.Pod).Name
			case *v1.Status:
				obj = e.Object.(*v1.Status).Message
			}
			t.Errorf("ResultChan should have been closed. Event: %s. Object: %s", e.Type, obj)
		}
	case <-time.After(wait.ForeverTestTimeout):
		t.Errorf("time out after waiting 1s on ResultChan")
	}
}

	w := store.watcher.createWatchChan(ctx, "/abc", 0, false, false, storage.Everything)
	// make resultChan and errChan blocking to ensure ordering.
	w.resultChan = make(chan watch.Event)
	w.errChan = make(chan error)
	// The event flow goes like:
	// - first we send an error, it should block on resultChan.
	// - Then we cancel ctx. The blocking on resultChan should be freed up
	//   and run() goroutine should return.
	var wg sync.WaitGroup

			defer cancel()
			resultChan, errChan := make(chan string, 1), make(chan error, 1)
			go func(resultChan chan string, errChan chan error, ctx context.Context, cniConfName, mountedCNINetDir string, chained bool) {
				result, err := getCNIConfigFilepath(ctx, cniConfName, mountedCNINetDir, chained)
				if err != nil {
					errChan <- err
					return
				}
				resultChan <- result

	}()

	// We expect all init events to be delivered.
	for i := 0; i < numObjects; i++ {
		<-w.ResultChan()
	}
	// We don't expect any other event to be delivered and thus
	// the ResultChan to be closed.
	result, ok := <-w.ResultChan()
	if ok {
		t.Errorf("unexpected event: %#v", result)
	}

	wg.Wait()
}

func TestTimeBucketWatchersBasic(t *testing.T) {

	d.RLock()
	defer d.RUnlock()
	return d.requestWatchProgressCounter
}

type dummyWatch struct {
	ch chan watch.Event
}

func (w *dummyWatch) ResultChan() <-chan watch.Event {
	return w.ch
}

func (w *dummyWatch) Stop() {
	close(w.ch)
}

func newDummyWatch() watch.Interface {
	return &dummyWatch{
		ch: make(chan watch.Event),
	}
}

			t.Errorf("expected %v, got %v", e, a)
		}
		for watchVersion, noxuWatch := range noxuWatchs {
			select {
			case watchEvent := <-noxuWatch.ResultChan():
				if e, a := watch.Added, watchEvent.Type; e != a {
					t.Errorf("expected %v, got %v", e, a)
					break
				}
				createdObjectMeta, err := meta.Accessor(watchEvent.Object)
				if err != nil {

			w, err := noxuNamespacedResourceClient.Watch(context.TODO(), metav1.ListOptions{})
			if err != nil {
				t.Errorf("unexpected error establishing watch: %v", err)
				return
			}
			for event := range w.ResultChan() {
				switch event.Type {
				case watch.Added, watch.Modified, watch.Deleted:
					// all expected
				default:
					t.Errorf("unexpected watch event: %#v", event)
				}
			}
		}(i)
	}

	})
	if err != nil {
		return nil, fmt.Errorf("failed to watch CSR %v", csr)
	}

	// Set a timeout
	timer := time.After(watchTimeout)
	for {
		select {
		case r := <-watcher.ResultChan():
			reqSigned := r.Object.(*cert.CertificateSigningRequest)
			if reqSigned.Status.Certificate != nil {
				return reqSigned.Status.Certificate, nil
			}
		case <-timer:

		ResourceVersion: "0",
		Predicate:       storage.Everything,
	})
	if err != nil {
		t.Fatal(err)
	}

	select {
	case _, ok := <-w.ResultChan():
		if ok {
			t.Error("ResultChan() should be closed")
		}
	case <-time.After(wait.ForeverTestTimeout):
		t.Errorf("timeout after %v", wait.ForeverTestTimeout)
	}
}