diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/configuration/DataFileConfigurationBuilderBase.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/configuration/DataFileConfigurationBuilderBase.java index 61589448b..aeb9a1b29 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/configuration/DataFileConfigurationBuilderBase.java +++ b/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 diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBase.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBase.java index a0995b9fb..879f55f44 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBase.java +++ b/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 diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/SingleFileAspectEngineBuilderBase.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/SingleFileAspectEngineBuilderBase.java index 4a0cf8667..84df5a4f9 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/SingleFileAspectEngineBuilderBase.java +++ b/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. *

- * 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. *

diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java index b21f1a317..398bfd9b9 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java +++ b/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 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 watchServices = + new IdentityHashMap<>(); private final Logger logger; private final boolean closeFutureFactory; private final List 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; diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBaseTest.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBaseTest.java index 196e6b9d3..efbd8ade3 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBaseTest.java +++ b/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) { diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceLoggingTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceLoggingTests.java index 15a2f9a1b..dcd972ae1 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceLoggingTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceLoggingTests.java @@ -279,7 +279,8 @@ public void UpdateOnStartup_AutoUpdateOff_StartsNoTimer() throws Exception { @Test public void FileSystemWatcher_LogsCreatingFileSystemWatcher() { - AspectEngineDataFileDefault file = newDataFile(false, true, false); + // The watcher needs no automatic updates. + AspectEngineDataFileDefault file = newDataFile(false, false, false); file.getConfiguration().setFileSystemWatcherEnabled(true); dataUpdate.registerDataFile(file); diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceTests.java index a1650c1da..b10b06a8a 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceTests.java @@ -374,12 +374,9 @@ public void DataUpdateService_Register_FileSystemWatcher() throws IOException { dataUpdate.registerDataFile(file); // Assert - // The automatic update switch governs the file system watcher. - if (autoUpdateEnabled) { - assertNotNull(config.getWatchKey()); - } else { - assertNull(config.getWatchKey()); - } + // The watcher runs on its own setting, whether or not automatic + // updates are on. + assertNotNull(config.getWatchKey()); } finally { // Make sure we tidy up the temp file. File tempFile = new File(tempData); @@ -398,7 +395,8 @@ public void DataUpdateService_UpdateFromWatcher() throws IOException, Interrupte // Arrange OnPremiseAspectEngine engine = mock(OnPremiseAspectEngine.class); DataFileConfigurationDefault config = new DataFileConfigurationDefault(); - config.setAutomaticUpdatesEnabled(true); + // The watcher needs no automatic updates. + config.setAutomaticUpdatesEnabled(false); config.setFileSystemWatcherEnabled(true); config.setPollingIntervalSeconds(1); @@ -424,8 +422,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable { }); // Configure a flag to be set when the watcher has applied the - // update. The polling timer's own check also completes, with no - // update found, and must not release the flag. + // update. final Semaphore completeFlag = new Semaphore(1); completeFlag.acquire(); dataUpdate.onUpdateComplete(new OnUpdateComplete() { @@ -1028,8 +1025,8 @@ public void DataUpdateService_CheckForUpdate_FileUpdated() { /** * A programmatic update compares the data file on disk whether or not - * Automatic Update from File is enabled, and with automatic update off - * no file system watcher exists whatever the watcher flag says. + * Automatic Update from File is enabled, here with automatic update off + * and the watcher flag at its default. */ @Test public void DataUpdateService_CheckForUpdate_FileUpdated_AutoUpdateOffWatcherFlagDefault() { diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java index b57a16ca9..b3e6d13c1 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java @@ -37,12 +37,20 @@ import org.mockito.Answers; import org.slf4j.Logger; +import java.io.ByteArrayInputStream; import java.io.File; import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.URL; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.WatchKey; import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -53,6 +61,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assume.assumeFalse; import static org.mockito.ArgumentMatchers.any; @@ -60,6 +69,7 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockingDetails; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.mockito.Mockito.withSettings; @@ -100,6 +110,10 @@ public class DataUpdateServiceWatchedFileTests { "File watchers are not well implemented in OS X, " + "so don't run this unit test as it is unlikely to pass."; + private static final String POLLER_MESSAGE = + "The watch service on OS X polls from a thread this test cannot " + + "pick out."; + @Test public void WatchedFileRefresh_PollingCarriesOn() throws Exception { assumeFalse(MAC_MESSAGE, isMac()); @@ -151,6 +165,283 @@ public void WatchedFileRefresh_PollingCarriesOn() throws Exception { } } + /** + * The file system watcher runs on its own setting. Watching a file + * downloads nothing, so the watcher does not wait for automatic updates + * to be on, and with its own setting off a replaced file is not loaded. + * Polling runs only with automatic updates on. + */ + @Test + public void WatcherRunsOnItsOwnSetting() throws Exception { + String[] names = { + "watcher on, automatic updates on", + "watcher on, automatic updates off", + "watcher off, automatic updates on"}; + boolean[] automatic = {true, false, true}; + boolean[] watching = {true, true, false}; + for (int i = 0; i < names.length; i++) { + checkWatcherSetting(names[i], automatic[i], watching[i]); + } + } + + private static void checkWatcherSetting( + String name, + boolean automatic, + boolean watching) throws Exception { + final List timers = new ArrayList<>(); + final AtomicReference watcher = new AtomicReference<>(); + FileWrapperFactory fileWrapperFactory = mock(FileWrapperFactory.class); + when(fileWrapperFactory.build(anyString())).then(invocation -> + new FileWrapperDefault((String) invocation.getArgument(0))); + when(fileWrapperFactory.getLastModified(anyString())).then( + invocation -> lastModified(invocation.getArgument(0))); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory, + futureFactory(timers, watcher)); + + OnPremiseAspectEngine engine = engine(); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + AspectEngineDataFileDefault file = + watchedFile(engine, dataFile, automatic, watching); + + try { + dataUpdate.registerDataFile(file); + assertEquals(name + ", a file system watcher was created", + watching, file.getConfiguration().getWatchKey() != null); + assertEquals(name + ", the watcher's task was scheduled", + watching, watcher.get() != null); + assertEquals(name + ", polling timers set", + automatic ? 1 : 0, timers.size()); + + // macOS watchers are too slow to wait for. The test holds the + // polling timers, so only a watcher can load the replaced file. + if (isMac() == false) { + Files.write(dataFile, "New".getBytes()); + Runnable task = watcher.get(); + long end = System.currentTimeMillis() + WAIT_FOR_EVENT_MILLIS; + while (task != null && refreshCount(engine) == 0 && + System.currentTimeMillis() < end) { + task.run(); + Thread.sleep(RETRY_MILLIS); + } + assertEquals(name + ", the replaced file was loaded", + watching ? 1 : 0, refreshCount(engine)); + } + } + finally { + cleanUp(file, directory, dataFile); + } + } + + /** + * Each watch service runs a thread of its own. Unregistering a watched + * data file, as closing its engine does, closes the watch service so its + * thread ends. + */ + @Test + public void Unregistering_ClosesTheWatchService() throws Exception { + assumeFalse(POLLER_MESSAGE, isMac()); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory(), + futureFactory(new ArrayList<>(), new AtomicReference<>())); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + AspectEngineDataFileDefault file = + watchedFile(engine(), dataFile, false, true); + + try { + Set before = watchServiceThreads(); + dataUpdate.registerDataFile(file); + assertEquals("Registering starts a watch service thread", + 1, waitForThreadsStartedSince(before, 1).size()); + + dataUpdate.unregisterDataFile(file); + assertEquals("Unregistering ends the watch service thread", + 0, waitForThreadsStartedSince(before, 0).size()); + } + finally { + cleanUp(file, directory, dataFile); + } + } + + /** + * A downloaded data file is copied into place with the watch paused. + * The watch then resumes on the data file's own watch service, so no + * other watch service is left running, and unregistering ends the one + * there is. + */ + @Test + public void DownloadedFile_WatchResumesOnTheSameWatchService() + throws Exception { + assumeFalse(POLLER_MESSAGE, isMac()); + final List timers = new ArrayList<>(); + HttpURLConnection connection = mock(HttpURLConnection.class); + when(connection.getResponseCode()) + .thenReturn(HttpURLConnection.HTTP_OK); + when(connection.getInputStream()).thenReturn( + new ByteArrayInputStream("New".getBytes())); + HttpClient httpClient = mock(HttpClient.class); + when(httpClient.connect(any(URL.class))).thenReturn(connection); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + httpClient, + fileWrapperFactory(), + timerFactory(timers)); + + OnPremiseAspectEngine engine = engine(); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + Path tempDirectory = Files.createTempDirectory("download"); + AspectEngineDataFileDefault file = + watchedFile(engine, dataFile, true, true); + file.setTempDataDirPath(tempDirectory.toString()); + file.getConfiguration().setDataUpdateUrl("http://www.test.com"); + file.getConfiguration().setDecompressContent(false); + file.getConfiguration().setVerifyMd5(false); + file.getConfiguration().setVerifyModifiedSince(false); + + try { + Set before = watchServiceThreads(); + dataUpdate.registerDataFile(file); + assertEquals("Registering starts a watch service thread", + 1, waitForThreadsStartedSince(before, 1).size()); + + // The polling timer downloads the new file. + timers.get(0).run(); + assertEquals("The downloaded file refreshes the engine", + 1, refreshCount(engine)); + assertNotNull("The watch resumes after the download", + file.getConfiguration().getWatchKey()); + assertEquals("The watch resumes on the same watch service", + 1, waitForThreadsStartedSince(before, 1).size()); + + dataUpdate.unregisterDataFile(file); + assertEquals("Unregistering ends the watch service thread", + 0, waitForThreadsStartedSince(before, 0).size()); + } + finally { + cleanUp(file, directory, dataFile); + tempDirectory.toFile().delete(); + } + } + + /** + * The watcher's task repeats, and a repeating task that fails is never + * run again. So while the watch is paused, as it is when a downloaded + * file is copied into place, the task does nothing and carries on. + * Unregistering the data file while the watch is paused still stops the + * task. + */ + @Test + public void WatchPaused_WatcherTaskCarriesOn() throws Exception { + final AtomicReference watcher = new AtomicReference<>(); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory(), + futureFactory(new ArrayList<>(), watcher)); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + AspectEngineDataFileDefault file = + watchedFile(engine(), dataFile, false, true); + + try { + dataUpdate.registerDataFile(file); + WatchKey watchKey = file.getConfiguration().getWatchKey(); + assertNotNull("A file system watcher was created", watchKey); + ScheduledFuture task = file.getPollFuture(); + + file.getConfiguration().setWatchKey(null); + watcher.get().run(); + file.getConfiguration().setWatchKey(watchKey); + watcher.get().run(); + + file.getConfiguration().setWatchKey(null); + dataUpdate.unregisterDataFile(file); + verify(task).cancel(true); + watchKey.cancel(); + } + finally { + cleanUp(file, directory, dataFile); + } + } + + /** + * The watcher's task polls the watch key, so with no watch key, here + * because the data file's folder does not exist, no task is scheduled. + */ + @Test + public void NoWatchKey_NoWatcherTask() throws Exception { + final AtomicReference watcher = new AtomicReference<>(); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory(), + futureFactory(new ArrayList<>(), watcher)); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("missing").resolve("data.dat"); + AspectEngineDataFileDefault file = + watchedFile(engine(), dataFile, false, true); + + try { + dataUpdate.registerDataFile(file); + assertNull("No watch key without the data file's folder", + file.getConfiguration().getWatchKey()); + assertNull("No watcher task without a watch key", + watcher.get()); + dataUpdate.unregisterDataFile(file); + } + finally { + directory.toFile().delete(); + } + } + + /** + * Registering a data file again that has no watch key, as it has while + * the watch is paused for a download, creates the watch key again and + * replaces the watcher's task rather than leave the old one running. + */ + @Test + public void RegisteredAgain_WatcherTaskReplaced() throws Exception { + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory(), + futureFactory(new ArrayList<>(), new AtomicReference<>())); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + AspectEngineDataFileDefault file = + watchedFile(engine(), dataFile, false, true); + + try { + dataUpdate.registerDataFile(file); + ScheduledFuture first = file.getPollFuture(); + assertNotNull("The watcher's task was scheduled", first); + + file.getConfiguration().setWatchKey(null); + dataUpdate.registerDataFile(file); + assertNotNull("The watch key is created again", + file.getConfiguration().getWatchKey()); + verify(first).cancel(false); + dataUpdate.unregisterDataFile(file); + } + finally { + cleanUp(file, directory, dataFile); + } + } + /** * A watched file that cannot be opened is tried, with a pause between * tries, until the wait ends. The update then reports @@ -351,6 +642,67 @@ private static FutureFactory futureFactory( return futureFactory; } + /** + * Keeps each one-off timer's task for the test to run itself. + */ + private static FutureFactory timerFactory(final List timers) { + FutureFactory futureFactory = mock(FutureFactory.class); + when(futureFactory.schedule(any(Runnable.class), anyLong())) + .then(invocation -> { + timers.add((Runnable) invocation.getArgument(0)); + return mock(ScheduledFuture.class); + }); + when(futureFactory.scheduleRepeating(any(Runnable.class), anyLong())) + .then(invocation -> mock(ScheduledFuture.class)); + return futureFactory; + } + + private static FileWrapperFactory fileWrapperFactory() { + FileWrapperFactory fileWrapperFactory = mock(FileWrapperFactory.class); + when(fileWrapperFactory.build(anyString())).then(invocation -> + new FileWrapperDefault((String) invocation.getArgument(0))); + when(fileWrapperFactory.getLastModified(anyString())).then( + invocation -> lastModified(invocation.getArgument(0))); + return fileWrapperFactory; + } + + /** + * The watch service threads running, found by the watch service's + * poller in each thread's stack. + */ + private static Set watchServiceThreads() { + Set threads = new HashSet<>(); + for (Map.Entry thread : + Thread.getAllStackTraces().entrySet()) { + if (Arrays.stream(thread.getValue()).anyMatch(frame -> + frame.getClassName().endsWith("WatchService$Poller"))) { + threads.add(thread.getKey()); + } + } + return threads; + } + + /** + * Wait for the watch service threads running that were not running + * before to number the count given, as a thread starts and ends a moment + * after its watch service is created or closed. Threads running before + * are left out, so one of those ending makes no difference. + */ + private static Set waitForThreadsStartedSince( + Set before, + int count) throws InterruptedException { + long end = System.currentTimeMillis() + WAIT_FOR_EVENT_MILLIS; + Set started = watchServiceThreads(); + started.removeAll(before); + while (started.size() != count && + System.currentTimeMillis() < end) { + Thread.sleep(RETRY_MILLIS); + started = watchServiceThreads(); + started.removeAll(before); + } + return started; + } + @SuppressWarnings("unchecked") private static OnPremiseAspectEngine engine() { @@ -361,9 +713,18 @@ private static AspectEngineDataFileDefault watchedFile( OnPremiseAspectEngine engine, Path dataFile) { + return watchedFile(engine, dataFile, true, true); + } + + private static AspectEngineDataFileDefault watchedFile( + OnPremiseAspectEngine engine, + Path dataFile, + boolean automatic, + boolean watching) { DataFileConfigurationDefault config = new DataFileConfigurationDefault(); - config.setAutomaticUpdatesEnabled(true); - config.setFileSystemWatcherEnabled(true); + config.setAutomaticUpdatesEnabled(automatic); + config.setFileSystemWatcherEnabled(watching); config.setPollingIntervalSeconds(1); config.setDataFilePath(dataFile.toString()); AspectEngineDataFileDefault file = new AspectEngineDataFileDefault();