Run the file system watcher on its own setting

pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/configuration/DataFileConfigurationBuilderBase.java

@@ -148,8 +148,9 @@ public TBuilder setAutoUpdate(boolean enabled) {
      * disk and automatically refresh the engine as soon as the file is updated.
      * This setting enables/disables that feature.
      *
-     * The {@link #setAutoUpdate(boolean)} feature must also be enabled in order
-     * for the file system watcher to work.
+     * The file system watcher runs on its own setting, whether
+     * {@link #setAutoUpdate(boolean)} is on or off, and needs no license key,
+     * as watching a file downloads nothing.
      * If the engine is built from a byte[] then this setting does nothing.
      * @param enabled  the cache configuration to use
      * @return this builder instance

pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBase.java

@@ -201,13 +201,27 @@ protected void preCreateEngine() {
             }
 
             if (dataFileConfig.getAutomaticUpdatesEnabled() ||
-                dataFileConfig.getUpdateOnStartup()) {
+                dataFileConfig.getUpdateOnStartup() ||
+                isWatched(dataFileConfig)) {
                 dataUpdateService.registerDataFile(dataFile);
             }
             dataFiles.add(dataFile);
         }
     }
 
+    /**
+     * Whether the data file is to be watched for a new file on disk. The
+     * file system watcher runs on its own setting and needs neither
+     * automatic updates nor a license key, only a data update service to run
+     * it and a data file on disk.
+     */
+    private boolean isWatched(DataFileConfiguration dataFileConfig) {
+        return dataUpdateService != null &&
+            dataFileConfig.getFileSystemWatcherEnabled() &&
+            dataFileConfig.getDataFilePath() != null &&
+            dataFileConfig.getDataFilePath().isEmpty() == false;
+    }
+
     /**
      * Refuse a data file configuration that needs a data update service or
      * a license key it does not have. Every data file is checked before any

pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/SingleFileAspectEngineBuilderBase.java

@@ -216,8 +216,9 @@ public TBuilder setAutoUpdate(boolean enabled) {
      * disk and automatically refresh the engine as soon as the file is updated.
      * This setting enables/disables that feature.
      * <p>
-     * The {@link #setAutoUpdate(boolean)} feature must also be enabled in order
-     * for the file system watcher to work.
+     * The file system watcher runs on its own setting, whether
+     * {@link #setAutoUpdate(boolean)} is on or off, and needs no license key,
+     * as watching a file downloads nothing.
      * <p>
      * If the engine is built from a byte[] then this setting does nothing.
      * <p>

pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java

@@ -92,6 +92,13 @@ public class DataUpdateServiceDefault implements DataUpdateService {
     long watchedFileOpenTimeoutMillis = WATCHED_FILE_OPEN_TIMEOUT_MILLIS;
 
     final List<AspectEngineDataFile> configurations = new ArrayList<>();
+    /**
+     * The watch service of each watched data file, held under the
+     * configuration lock. A watch service runs a thread of its own, so it is
+     * closed when its data file is unregistered.
+     */
+    private final Map<AspectEngineDataFile, WatchService> watchServices =
+            new IdentityHashMap<>();
     private final Logger logger;
     private final boolean closeFutureFactory;
     private final List<OnUpdateComplete> onUpdateCompleteList = new ArrayList<>();
@@ -548,11 +555,10 @@ public void run() {
                 }
             }
 
-            // If file system watcher is enabled then set it up. The
-            // automatic update switch governs the watcher as well as the
-            // polling timer.
-            if (aspectDataFile.getAutomaticUpdatesEnabled() &&
-                    aspectDataFile.getConfiguration().getFileSystemWatcherEnabled() &&
+            // If file system watcher is enabled then set it up. The watcher
+            // runs on its own setting. Watching a file downloads nothing, so
+            // it needs neither automatic updates nor a license key.
+            if (aspectDataFile.getConfiguration().getFileSystemWatcherEnabled() &&
                     aspectDataFile.getConfiguration().getWatchKey() == null &&
                     aspectDataFile.getDataFilePath() != null &&
                     aspectDataFile.getDataFilePath().isEmpty() == false) {
@@ -568,9 +574,8 @@ public void run() {
                         throw new IllegalStateException("Cannot find directory to watch");
                     }
                     logger.debug("Getting a watcher using {}", aspectDataFilePath);
-                    WatchService watcher = aspectDataFilePath
-                            .getFileSystem()
-                            .newWatchService();
+                    WatchService watcher =
+                            watchService(aspectDataFile, aspectDataFilePath);
                     logger.debug("Setting a watcher for {}", aspectDataFilePath.getParent());
                     aspectDataFile.getConfiguration().setWatchKey(aspectDataFilePath.getParent().register(
                             watcher,
@@ -579,33 +584,49 @@ public void run() {
                     logger.error("File watcher for '{}' could not be initialised. " + e.getMessage(),
                             aspectDataFile.getDataFilePath(), e);
                 }
-                try {
-                    long pollingInterval =
-                            aspectDataFile.getConfiguration().getPollingIntervalSeconds() * 1000L;
-                    logger.debug("Setting poll future {} millis", pollingInterval);
-                    aspectDataFile.setPollFuture(futureFactory.scheduleRepeating(
-                            new Runnable() {
-                                @Override
-                                public void run() {
-                                    logger.debug("Watcher is running");
-                                    for (WatchEvent<?> event :
-                                            aspectDataFile.getConfiguration().getWatchKey().pollEvents()) {
-                                        if (event.kind() == StandardWatchEventKinds.ENTRY_MODIFY) {
-                                            if (((Path) event.context()).endsWith(aspectDataFilePath.getFileName())) {
-                                                dataFileUpdated(aspectDataFile);
+                // The watcher's task polls the watch key, so it is scheduled
+                // only when there is one, and replaces any task scheduled
+                // before rather than run beside it.
+                if (aspectDataFile.getConfiguration().getWatchKey() != null) {
+                    try {
+                        if (aspectDataFile.getPollFuture() != null) {
+                            aspectDataFile.getPollFuture().cancel(false);
+                        }
+                        long pollingInterval =
+                                aspectDataFile.getConfiguration().getPollingIntervalSeconds() * 1000L;
+                        logger.debug("Setting poll future {} millis", pollingInterval);
+                        aspectDataFile.setPollFuture(futureFactory.scheduleRepeating(
+                                new Runnable() {
+                                    @Override
+                                    public void run() {
+                                        logger.debug("Watcher is running");
+                                        // The watch is paused while a downloaded
+                                        // file is copied into place. A repeating
+                                        // task that fails never runs again, so
+                                        // this run does nothing.
+                                        WatchKey watchKey =
+                                                aspectDataFile.getConfiguration().getWatchKey();
+                                        if (watchKey == null) {
+                                            return;
+                                        }
+                                        for (WatchEvent<?> event : watchKey.pollEvents()) {
+                                            if (event.kind() == StandardWatchEventKinds.ENTRY_MODIFY) {
+                                                if (((Path) event.context()).endsWith(aspectDataFilePath.getFileName())) {
+                                                    dataFileUpdated(aspectDataFile);
+                                                }
                                             }
                                         }
+                                        // make sure to reset the watch key or we are not called again
+                                        watchKey.reset();
                                     }
-                                    // make sure to reset the watch key or we are not called again
-                                    aspectDataFile.getConfiguration().getWatchKey().reset();
-                                }
-                            },
-                            pollingInterval));
-                    logger.debug("Poll check scheduled for {} mins",
-                            aspectDataFile.getPollFuture().getDelay(TimeUnit.MINUTES));
-                } catch (Exception e) {
-                    logger.error("Could not schedule watcher for {}.",
-                            aspectDataFile.getDataFilePath(), e);
+                                },
+                                pollingInterval));
+                        logger.debug("Poll check scheduled for {} mins",
+                                aspectDataFile.getPollFuture().getDelay(TimeUnit.MINUTES));
+                    } catch (Exception e) {
+                        logger.error("Could not schedule watcher for {}.",
+                                aspectDataFile.getDataFilePath(), e);
+                    }
                 }
             }
 
@@ -630,11 +651,13 @@ public void unregisterDataFile(AspectEngineDataFile dataFile) {
                 }
                 AspectEngineDataFileDefault aspectDataFile =
                         (AspectEngineDataFileDefault) dataFile;
+                // The watcher's task is stopped even while the watch is
+                // paused for a download.
+                if (aspectDataFile.getPollFuture() != null) {
+                    aspectDataFile.getPollFuture().cancel(true);
+                    aspectDataFile.setPollFuture(null);
+                }
                 if (aspectDataFile.getConfiguration().getWatchKey() != null) {
-                    if (aspectDataFile.getPollFuture() != null) {
-                        aspectDataFile.getPollFuture().cancel(true);
-                        aspectDataFile.setPollFuture(null);
-                    }
                     aspectDataFile.getConfiguration().getWatchKey().cancel();
                     aspectDataFile.getConfiguration().setWatchKey(null);
                 }
@@ -645,12 +668,57 @@ public void unregisterDataFile(AspectEngineDataFile dataFile) {
                     }
                 }
             }
+            closeWatchService(dataFile);
         }
         if (dataFile != null) {
             dataFile.setDataUpdateService(null);
         }
     }
 
+    /**
+     * Get the data file's watch service, creating it the first time. The
+     * data file keeps the one watch service until it is unregistered, so the
+     * watch resumes on it after a download.
+     *
+     * @param dataFile the data file to watch
+     * @param path the absolute path of the data file
+     * @return the data file's watch service
+     * @throws IOException if a watch service cannot be created
+     */
+    private WatchService watchService(
+            AspectEngineDataFile dataFile,
+            Path path) throws IOException {
+        synchronized (configLock) {
+            WatchService watcher = watchServices.get(dataFile);
+            if (watcher == null) {
+                watcher = path.getFileSystem().newWatchService();
+                watchServices.put(dataFile, watcher);
+            }
+            return watcher;
+        }
+    }
+
+    /**
+     * Close the data file's watch service, if it has one, which ends the
+     * watch service's thread.
+     *
+     * @param dataFile the data file no longer watched
+     */
+    private void closeWatchService(AspectEngineDataFile dataFile) {
+        WatchService watcher;
+        synchronized (configLock) {
+            watcher = watchServices.remove(dataFile);
+        }
+        if (watcher != null) {
+            try {
+                watcher.close();
+            } catch (IOException e) {
+                logger.warn("Could not close the file watcher for '{}'",
+                        dataFile.getDataFilePath(), e);
+            }
+        }
+    }
+
     /**
      * Get the polling interval for a data file plus a random number of
      * seconds, no greater than the configured maximum randomisation.
@@ -1137,14 +1205,20 @@ private AutoUpdateStatus checkForUpdateFromUrl(
                             dataFile.getDataFilePath(), ex);
                     result = AutoUpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME;
                 }
-                // Make sure to enable the file watcher again if needed.
+                // Make sure to enable the file watcher again if needed. The
+                // watch resumes on the data file's own watch service, unless
+                // the data file was unregistered during the download.
                 if (watchKey != null) {
                     final Path filePath =
                             Paths.get(dataFile.getDataFilePath()).toAbsolutePath();
-                    WatchService watcher = filePath.getFileSystem().newWatchService();
-                    watchKey = filePath.getParent().register(watcher,
-                            StandardWatchEventKinds.ENTRY_MODIFY);
-                    dataFile.getConfiguration().setWatchKey(watchKey);
+                    synchronized (configLock) {
+                        if (configurations.contains(dataFile)) {
+                            watchKey = filePath.getParent().register(
+                                    watchService(dataFile, filePath),
+                                    StandardWatchEventKinds.ENTRY_MODIFY);
+                            dataFile.getConfiguration().setWatchKey(watchKey);
+                        }
+                    }
                 }
             }
             return result;

pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBaseTest.java

@@ -26,7 +26,11 @@
 import fiftyone.pipeline.engines.Constants;
 import fiftyone.pipeline.engines.configuration.DataFileConfiguration;
 import fiftyone.pipeline.engines.configuration.DataFileConfigurationDefault;
+import fiftyone.pipeline.engines.data.AspectEngineDataFile;
 import fiftyone.pipeline.engines.services.DataUpdateService;
+import fiftyone.pipeline.engines.services.DataUpdateServiceDefault;
+import fiftyone.pipeline.engines.services.HttpClient;
+import fiftyone.pipeline.engines.services.update.FutureFactory;
 import org.junit.After;
 import org.junit.Assume;
 import org.junit.Before;
@@ -42,14 +46,18 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ScheduledFuture;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 @SuppressWarnings("rawtypes")
 public class OnPremiseAspectEngineBuilderBaseTest {
@@ -177,6 +185,94 @@ protected boolean isLicenseKeyRequired(
         verify(updateService, never()).registerDataFile(any());
     }
 
+    /**
+     * The file system watcher runs on its own setting and needs no license
+     * key. An engine whose update source needs a key it does not have is
+     * built with automatic updates and the update on start-up off, and its
+     * data file is still registered to be watched. With the watcher off as
+     * well, nothing is registered.
+     */
+    @Test
+    public void WatcherAlone_RegistersTheDataFile() throws Exception {
+        String[] names = {
+            "watcher on, automatic updates off, no license key",
+            "watcher off, automatic updates off, no license key"};
+        boolean[] watching = {true, false};
+        for (int i = 0; i < names.length; i++) {
+            checkWatcherAlone(names[i], watching[i]);
+        }
+    }
+
+    private static void checkWatcherAlone(String name, boolean watching)
+        throws Exception {
+        FutureFactory futureFactory = mock(FutureFactory.class);
+        when(futureFactory.scheduleRepeating(any(Runnable.class), anyLong()))
+            .then(invocation -> mock(ScheduledFuture.class));
+        DataUpdateService updateService = new DataUpdateServiceDefault(
+            null, mock(HttpClient.class), null, futureFactory);
+        OnPremiseAspectEngineBuilderBase builder =
+            keyRequiredBuilder(updateService);
+
+        Path directory = Files.createTempDirectory("watched");
+        Path dataFile = directory.resolve("data.dat");
+        Files.write(dataFile, "Data".getBytes());
+        DataFileConfigurationDefault config = new DataFileConfigurationDefault();
+        config.setIdentifier("Watched");
+        config.setDataFilePath(dataFile.toString());
+        config.setAutomaticUpdatesEnabled(false);
+        config.setUpdateOnStartup(false);
+        config.setFileSystemWatcherEnabled(watching);
+        builder.addDataFile(config);
+
+        try {
+            builder.buildEngine();
+            AspectEngineDataFile file =
+                (AspectEngineDataFile) builder.dataFiles.get(0);
+            assertEquals(name + ", the data file is registered",
+                watching, file.getIsRegistered());
+            assertEquals(name + ", a file system watcher was created",
+                watching, config.getWatchKey() != null);
+        }
+        finally {
+            for (Object file : builder.dataFiles) {
+                updateService.unregisterDataFile((AspectEngineDataFile) file);
+            }
+            if (config.getWatchKey() != null) {
+                config.getWatchKey().cancel();
+            }
+            directory.toFile().deleteOnExit();
+            if (dataFile.toFile().delete() == false) {
+                dataFile.toFile().deleteOnExit();
+            }
+            directory.toFile().delete();
+        }
+    }
+
+    /**
+     * A builder whose update source serves nothing without a license key.
+     */
+    private static OnPremiseAspectEngineBuilderBase keyRequiredBuilder(
+        DataUpdateService updateService) {
+        return new OnPremiseAspectEngineBuilderBase(null, updateService) {
+            @Override
+            public OnPremiseAspectEngineBuilderBase setPerformanceProfile(
+                Constants.PerformanceProfiles profile) {
+                return this;
+            }
+
+            @Override
+            protected AspectEngine newEngine(List properties) {
+                return null;
+            }
+
+            @Override
+            protected boolean isLicenseKeyRequired(
+                DataFileConfiguration dataFileConfig) {
+                return true;
+            }
+        };
+    }
+
     private static DataFileConfiguration autoUpdateConfiguration(
         String identifier,
         String licenseKey) {