}

    @Override
    protected LinuxFileWatcher createFileWatcher(BlockingQueue<FileWatchEvent> fileEvents) throws InterruptedException {
        return fileEventFunctions.newWatcher(fileEvents)
            .start();
    }

    @Override
    protected FileWatcherUpdater createFileWatcherUpdater(LinuxFileWatcher watcher, FileWatcherProbeRegistry probeRegistry, WatchableHierarchies watchableHierarchies) {

	return w.watcher.Add(path)
}

func (w *fsnotifyWatcher) Init(eventHandler FSEventHandler, errorHandler FSErrorHandler) error {
	var err error
	w.watcher, err = fsnotify.NewWatcher()
	if err != nil {
		return err
	}

	w.eventHandler = eventHandler
	w.errorHandler = errorHandler
	return nil
}

func (w *fsnotifyWatcher) Run() {
	go func() {
		defer w.watcher.Close()

    }

    @Override
    protected WindowsFileWatcher createFileWatcher(BlockingQueue<FileWatchEvent> fileEvents) throws InterruptedException {
        return fileEventFunctions.newWatcher(fileEvents)
            .withBufferSize(BUFFER_SIZE)
            .start();
    }

    @Override
    protected FileWatcherUpdater createFileWatcherUpdater(
        WindowsFileWatcher watcher,

    }

    @Override
    protected OsxFileWatcher createFileWatcher(BlockingQueue<FileWatchEvent> fileEvents) throws InterruptedException {
        return fileEventFunctions.newWatcher(fileEvents)
            // TODO Figure out a good value for this
            .withLatency(20, TimeUnit.MICROSECONDS)
            .start();
    }

    @Override

func createTestController(t *testing.T) (*Controller, *atomic.Pointer[error]) {
	c := kube.NewFakeClient()
	revision := "default"
	ns := "default"
	watcher := keycertbundle.NewWatcher()
	watcher.SetAndNotify(nil, nil, caBundle0)
	control := newController(Options{
		WatchedNamespace: ns,
		CABundleWatcher:  watcher,
		Revision:         revision,
		ServiceName:      "istiod",
	}, c)

	CertPem  []byte
	KeyPem   []byte
	CABundle []byte
}

type Watcher struct {
	mutex     sync.RWMutex
	bundle    KeyCertBundle
	watcherID int32
	watchers  map[int32]chan struct{}
}

func NewWatcher() *Watcher {
	return &Watcher{
		watchers: make(map[int32]chan struct{}),
	}
}

// AddWatcher returns channel to receive the updated items.
func (w *Watcher) AddWatcher() (int32, chan struct{}) {

		s.updates <- kubetypes.PodUpdate{Pods: []*v1.Pod{}, Op: kubetypes.SET, Source: kubetypes.FileSource}
		return &retryableError{"path does not exist, ignoring"}
	}

	w, err := fsnotify.NewWatcher()
	if err != nil {
		return fmt.Errorf("unable to create inotify: %v", err)
	}
	defer w.Close()

	err = w.Add(s.path)
	if err != nil {
		return fmt.Errorf("unable to create inotify for path %q: %v", s.path, err)

func (w *Watcher) Close() {
	_ = w.watcher.Close()
}

// Creates a file watcher that watches for any changes to the directory
func CreateFileWatcher(paths ...string) (*Watcher, error) {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, fmt.Errorf("watcher create: %v", err)
	}

	fileModified, errChan := make(chan struct{}), make(chan error)
	go watchFiles(watcher, fileModified, errChan)

			operationexecutor.NewOperationGenerator(
				recorder,
			),
		),
		loopSleepDuration,
		dsw,
		asw,
	)

	pm := &pluginManager{
		desiredStateOfWorldPopulator: pluginwatcher.NewWatcher(
			sockDir,
			dsw,
		),
		reconciler:          reconciler,
		desiredStateOfWorld: dsw,
		actualStateOfWorld:  asw,
	}
	return pm
}

// pluginManager implements the PluginManager interface

	if !cfg.Enabled {
		return nil, nil
	}
	cli, err := clientv3.New(cfg.Config)
	if err != nil {
		return nil, err
	}
	cli.KV = namespace.NewKV(cli.KV, cfg.PathPrefix)
	cli.Watcher = namespace.NewWatcher(cli.Watcher, cfg.PathPrefix)
	cli.Lease = namespace.NewLease(cli.Lease, cfg.PathPrefix)
	return cli, nil
}

func parseEndpoints(endpoints string) ([]string, bool, error) {