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

		f:        f,
	}
	go fw.loop()
	return fw
}

type filteredWatch struct {
	incoming Interface
	result   chan Event
	f        FilterFunc
}

// ResultChan returns a channel which will receive filtered events.
func (fw *filteredWatch) ResultChan() <-chan Event {
	return fw.result
}

// Stop stops the upstream watch, which will eventually stop this watch.
func (fw *filteredWatch) Stop() {
	fw.incoming.Stop()
}

	}()

	// 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) {

		// error reporting might block forever.
		// Therefore a dedicated stop channel is used to resolve this blocking.
		done: make(chan struct{}),
	}
	go sw.receive()
	return sw
}

// ResultChan implements Interface.
func (sw *StreamWatcher) ResultChan() <-chan Event {
	return sw.result
}

// Stop implements Interface.
func (sw *StreamWatcher) Stop() {
	// Call Close() exactly once by locking and setting a flag.
	sw.Lock()

	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 {

type broadcasterWatcher struct {
	result  chan Event
	stopped chan struct{}
	stop    sync.Once
	id      int64
	m       *Broadcaster
}

// ResultChan returns a channel to use for waiting on events.
func (mw *broadcasterWatcher) ResultChan() <-chan Event {
	return mw.result
}

// Stop stops watching and removes mw from its list.
// It will block until the watcher stop request is actually executed

			case <-stopChannel:
				return
			case eventEvent, ok := <-eventWatch.ResultChan():
				if !ok {
					return
				}
				eventCh <- eventEvent
			}
		}
	}()

	go func() {
		defer podWatch.Stop()
		defer wg.Done()
		for {
			select {
			case <-stopChannel:
				return

			case podEvent, ok := <-podWatch.ResultChan():
				if !ok {
					return
				}
				eventCh <- podEvent

			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)
	}

	w.WriteHeader(http.StatusOK)
	flusher.Flush()

	kind := s.Scope.Kind
	watchEncoder := newWatchEncoder(req.Context(), kind, s.EmbeddedEncoder, s.Encoder, framer)
	ch := s.Watching.ResultChan()
	done := req.Context().Done()

	for {
		select {
		case <-s.ServerShuttingDownCh:
			// the server has signaled that it is shutting down (not accepting