// How large to make watcher's channel.
	watchQueueLength int
	// If one of the watch channels is full, don't wait for it to become empty.
	// Instead just deliver it to the watchers that do have space in their
	// channels and move on to the next event.
	// It's more fair to do this on a per-watcher basis than to do it on the
	// "incoming" channel, which would allow one slow watcher to prevent all

	}
	watchers0 := watchers.popExpiredWatchersThreadUnsafe()
	if len(watchers0) != 0 {
		t.Errorf("unexpected bucket size: %#v", watchers0)
	}

	clock.Step(10 * time.Second)
	watchers1 := watchers.popExpiredWatchersThreadUnsafe()
	if len(watchers1) != 1 || len(watchers1[0]) != 1 {
		t.Errorf("unexpected bucket size: %v", watchers1)
	}
	watchers1 = watchers.popExpiredWatchersThreadUnsafe()
	if len(watchers1) != 0 {

        private final LinuxFileWatcher watcher;
        private final WatchableHierarchies watchableHierarchies;

        public LinuxMovedDirectoryHandler(LinuxFileWatcher watcher, WatchableHierarchies watchableHierarchies) {
            this.watcher = watcher;
            this.watchableHierarchies = watchableHierarchies;
        }

        @Override

		return nil, nil
	}

	// If the injection config exists either locally or remotely, we will set up injection.
	var watcher inject.Watcher
	if _, err := os.Stat(filepath.Join(injectPath, "config")); !os.IsNotExist(err) {
		configFile := filepath.Join(injectPath, "config")
		valuesFile := filepath.Join(injectPath, "values")
		watcher, err = inject.NewFileWatcher(configFile, valuesFile)
		if err != nil {
			return nil, err
		}

        try {
            watcher.shutdown();
            if (!watcher.awaitTermination(5, TimeUnit.SECONDS)) {
                throw new RuntimeException("Watcher did not terminate within 5 seconds");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new InterruptedIOException("Awaiting termination of watcher was interrupted");
        } finally {

	}
}

// serveWatch will serve a watch response according to the watcher and watchServer.
// Before watchServer.HandleHTTP, an error may occur like k8s.io/apiserver/pkg/endpoints/handlers/watch.go#serveWatch does.
func serveWatch(watcher watch.Interface, watchServer *handlers.WatchServer, preServeErr error) http.HandlerFunc {
	return func(w http.ResponseWriter, req *http.Request) {
		defer watcher.Stop()

		if preServeErr != nil {

)

// possible states of the cache watcher
const (
	// cacheWatcherWaitingForBookmark indicates the cacher
	// is waiting for a bookmark event with a specific RV set
	cacheWatcherWaitingForBookmark = iota

	// cacheWatcherBookmarkReceived indicates that the cacher
	// has received a bookmark event with required RV
	cacheWatcherBookmarkReceived

	// cacheWatcherBookmarkSent indicates that the cacher

	err := prober.watcher.Init(func(event fsnotify.Event) {
		if err := prober.handleWatchEvent(event); err != nil {
			klog.Errorf("Flexvolume prober watch: %s", err)
		}
	}, func(err error) {
		klog.Errorf("Received an error from watcher: %s", err)
	})
	if err != nil {
		return fmt.Errorf("error initializing watcher: %s", err)
	}

	if err := prober.addWatchRecursive(prober.pluginDir); err != nil {

	"istio.io/istio/pkg/test"
	"istio.io/istio/pkg/test/util/retry"
)

func TestNamespaceController(t *testing.T) {
	client := kube.NewFakeClient()
	t.Cleanup(client.Shutdown)
	watcher := keycertbundle.NewWatcher()
	caBundle := []byte("caBundle")
	watcher.SetAndNotify(nil, nil, caBundle)
	meshWatcher := mesh.NewTestWatcher(&meshconfig.MeshConfig{})
	stop := test.NewStop(t)
	discoveryNamespacesFilter := filter.NewDiscoveryNamespacesFilter(

			inf.Start(s.started)
		}
	}
}

type delayedFilter struct {
	Watcher  kubetypes.CrdWatcher
	Resource schema.GroupVersionResource
}

func (d *delayedFilter) HasSynced() bool {
	return d.Watcher.HasSynced()
}

func (d *delayedFilter) KnownOrCallback(f func(stop <-chan struct{})) bool {
	return d.Watcher.KnownOrCallback(d.Resource, f)
}