defer cacher.stopWg.Done()
		defer cacher.terminateAllWatchers()
		wait.Until(
			func() {
				if !cacher.isStopped() {
					cacher.startCaching(stopCh)
				}
			}, time.Second, stopCh,
		)
	}()

	return cacher, nil
}

func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
	// The 'usable' lock is always 'RLock'able when it is safe to use the cache.

        addSnapshot(snapshotRegularFile(fileInWatchableHierarchies))
        then:
        1 * watcher.startWatching({ equalIgnoringOrder(it, [probeRegistry.getProbeDirectory(file("first"))]) })
        1 * watcher.startWatching({ equalIgnoringOrder(it, [fileInWatchableHierarchies.parentFile]) })
        1 * watcher.startWatching({ equalIgnoringOrder(it, [watchableHierarchies[0]]) })
        0 * _

        when:

        when:
        addSnapshotInWatchableHierarchy(secondDir)
        then:
        1 * watcher.startWatching({ equalIgnoringOrder(it, [secondDir]) })
        0 * _

        when:
        def snapshot = addSnapshotInWatchableHierarchy(directoryWithinFirst)
        then:
        1 * watcher.startWatching({ equalIgnoringOrder(it, [firstDir]) })
        0 * _

        when:

        then:
        watchableHierarchies.each { watchableHierarchy ->
            1 * watcher.startWatching({ equalIgnoringOrder(it, [watchableHierarchy]) })
            ifNonHierarchical * watcher.startWatching({ equalIgnoringOrder(it, [directoryContainingSnapshot(watchableHierarchy)]) })
            ifNonHierarchical * watcher.startWatching({ equalIgnoringOrder(it, [probeRegistry.getProbeDirectory(watchableHierarchy)]) })
        }

	//
	// when this variable is set to true,
	// a special annotation will be added
	// to the bookmark event.
	//
	// note that we decided to extend the event
	// struct field to eliminate contention
	// between startWatching and processEvent
	isInitialEventsEndBookmark bool
}

// parseKV converts a KeyValue retrieved from an initial sync() listing to a synthetic isCreated event.
func parseKV(kv *mvccpb.KeyValue) *event {
	return &event{

                        if (couldDetectUnsupportedFileSystems) {
                            context.setStatus("Starting file system watching");
                            newRoot = startWatching(currentRoot, watchMode, unsupportedFileSystems);
                        } else {
                            newRoot = currentRoot.empty();
                        }

        if (!hierarchiesToStartWatching.isEmpty()) {
            hierarchiesToStartWatching.forEach(locationToWatchValidator::validateLocationToWatch);
            fileWatcher.startWatching(hierarchiesToStartWatching);
        }

        LOGGER.debug("Watching {} directory hierarchies to track changes", watchedHierarchies.size());
    }

    @Override

		return true
	}
	return false
}

func (wc *watchChan) run(initialEventsEndBookmarkRequired, forceInitialEvents bool) {
	watchClosedCh := make(chan struct{})
	go wc.startWatching(watchClosedCh, initialEventsEndBookmarkRequired, forceInitialEvents)

	var resultChanWG sync.WaitGroup
	resultChanWG.Add(1)
	go wc.processEvent(&resultChanWG)

	select {
	case err := <-wc.errChan:

                    LOGGER.debug("Couldn't stop watching directories: {}", directoriesToStopWatching);
                }
            }
            if (!directoriesToStartWatching.isEmpty()) {
                fileWatcher.startWatching(directoriesToStartWatching);
            }
        } catch (NativeException e) {
            if (e.getMessage().contains("Already watching path: ")) {