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 007b2121..f366dbfb 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 @@ -226,6 +226,15 @@ private AutoUpdateStatus checkForUpdate(Object state, boolean manualUpdate) { // yet been refreshed. logger.debug("Calling updatedFileAvailable"); result = updatedFileAvailable(dataFile); + } else if (result == AUTO_UPDATE_SUCCESS) { + // The update was applied in memory. Clear the timer, as + // updatedFileAvailable does, so a new one is registered. + synchronized (futureLock) { + if (dataFile.getFuture() != null) { + dataFile.getFuture().cancel(true); + dataFile.setFuture(null); + } + } } else { // No update available. // If this was a manual call to update then do nothing. @@ -249,9 +258,15 @@ public void run() { } } } - if (manualUpdate == false) { + if (manualUpdate == false || + ((result == AUTO_UPDATE_SUCCESS || + result == AUTO_UPDATE_REFRESH_FAILED) && + dataFile.getIsRegistered())) { // Re-register the engine with the data update service // so it knows when next set of data should be available. + // A manual update of a registered data file does this too, + // whether or not the refresh worked, as the refresh clears + // the timer, so polling resumes after it. registerDataFile(dataFile); } @@ -565,6 +580,9 @@ public void unregisterDataFile(AspectEngineDataFile dataFile) { } } } + if (dataFile != null) { + dataFile.setDataUpdateService(null); + } } /** @@ -805,7 +823,8 @@ private AutoUpdateStatus checkForUpdateFromUrl(AspectEngineDataFile dataFile) { dataFile.getConfiguration().setData(reader.toByteArray()); } } - return memoryDownloadResult; + // The engine, or the configuration, now holds the new data. + return AUTO_UPDATE_SUCCESS; } catch (Exception ex) { logger.error("An error occurred when applying a data update to engine '{}'" , Objects.nonNull(dataFile.getEngine())? 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 52176409..4ed689fc 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 @@ -45,6 +45,7 @@ import org.mockito.ArgumentCaptor; import org.mockito.ArgumentMatcher; import org.mockito.internal.matchers.LessOrEqual; +import org.mockito.invocation.Invocation; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.slf4j.LoggerFactory; @@ -61,6 +62,7 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.nio.file.StandardCopyOption; +import java.util.ArrayList; import java.util.Arrays; import java.util.Calendar; import java.util.Collection; @@ -72,6 +74,7 @@ import static fiftyone.pipeline.engines.services.DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_HTTPS_ERR; import static fiftyone.pipeline.engines.services.DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_NOT_NEEDED; +import static fiftyone.pipeline.engines.services.DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_SUCCESS; import static org.junit.Assert.*; import static org.junit.Assume.assumeFalse; import static org.mockito.Mockito.*; @@ -415,14 +418,17 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable { } }); - // Configure a flag to be set when processing - // is complete. + // 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. final Semaphore completeFlag = new Semaphore(1); completeFlag.acquire(); dataUpdate.onUpdateComplete(new OnUpdateComplete() { @Override public void call(Object sender, DataUpdateService.DataUpdateCompleteArgs args) { - completeFlag.release(); + if (args.getStatus() == AUTO_UPDATE_SUCCESS) { + completeFlag.release(); + } } }); @@ -1229,6 +1235,297 @@ public boolean matches(byte[] argument) { verify(futureFactory, times(1)).schedule(any(Runnable.class), anyLong()); } + /** + * A check that finds an update for a data file with no file path + * applies it in memory, so it refreshes the engine once, with the + * downloaded data, and does not refresh it again. + */ + @Test + public void DataUpdateService_CheckForUpdate_InMemory_RefreshesOnce() + throws IOException { + // Arrange + configureNoFileSystem(); + when(httpClientConnection.getResponseCode()) + .thenReturn(HttpURLConnection.HTTP_OK); + when(httpClientConnection.getResponseMessage()).thenReturn(null); + when(httpClientConnection.getHeaderField("Content-MD5")) + .thenReturn("08527dcbdd437e7fa6c084423d06dba6"); + when(httpClientConnection.getInputStream()).thenReturn( + DataUpdateServiceTests.class.getClassLoader() + .getResourceAsStream("file.gz")); + + OnPremiseAspectEngine engine = + mock(OnPremiseAspectEngine.class); + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setDataUpdateUrl("http://www.test.com"); + config.setVerifyMd5(true); + config.setDecompressContent(true); + config.setVerifyModifiedSince(false); + + AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); + file.setEngine(engine); + file.setConfiguration(config); + when(engine.getDataFileMetaData()).thenReturn(file); + + // Act + DataUpdateService.AutoUpdateStatus status = + dataUpdate.checkForUpdate(engine); + + // Assert + assertEquals(AUTO_UPDATE_SUCCESS, status); + List refreshes = new ArrayList<>(); + for (Invocation invocation : mockingDetails(engine).getInvocations()) { + if (invocation.getMethod().getName().equals("refreshData")) { + refreshes.add(invocation.getMethod().toGenericString()); + } + } + assertEquals( + "The engine was not refreshed exactly once: " + refreshes, + 1, + refreshes.size()); + verify(engine, times(1)).refreshData(any(), any(byte[].class)); + } + + /** + * A manual check that applies an update from the data file to a data + * file registered for automatic updates sets the next update timer, as + * a check the timer started does. + */ + @Test + public void DataUpdateService_CheckForUpdate_FileUpdated_TimerSetAgain() { + // Arrange + OnPremiseAspectEngine engine = + mock(OnPremiseAspectEngine.class); + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setAutomaticUpdatesEnabled(autoUpdateEnabled); + config.setFileSystemWatcherEnabled(false); + + AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); + file.setEngine(engine); + file.setConfiguration(config); + when(engine.getDataFileMetaData()).thenReturn(file); + configureFileUpdate(engine); + + dataUpdate.registerDataFile(file); + int timers = autoUpdateEnabled ? 1 : 0; + verify(futureFactory, times(timers)).schedule( + any(Runnable.class), + anyLong()); + + ScheduledFuture pending = file.getFuture(); + + // Act + DataUpdateService.AutoUpdateStatus status = + dataUpdate.checkForUpdate(engine); + + // Assert + assertEquals(AUTO_UPDATE_SUCCESS, status); + verify(engine, times(1)).refreshData((String) any()); + verify(futureFactory, times(2 * timers)).schedule( + any(Runnable.class), + anyLong()); + assertEquals( + "The next update timer was not set after the manual update.", + autoUpdateEnabled, + file.getFuture() != null); + + if (autoUpdateEnabled) { + // The pending timer is cancelled, so only the new one is left. + assertTrue( + "The timer pending before the manual update was not cancelled.", + pending.isCancelled()); + assertNotSame(pending, file.getFuture()); + } + } + + /** + * A manual check whose refresh from the data file fails still leaves + * the next update timer set, as a check the timer started does, so + * polling resumes. + */ + @Test + public void DataUpdateService_CheckForUpdate_FileRefreshFails_TimerSetAgain() { + // updatedFileAvailable warns on each of its 10 tries. + ignoreWranings = 10; + + // Arrange + OnPremiseAspectEngine engine = + mock(OnPremiseAspectEngine.class); + doThrow(new IllegalStateException("The refresh failed.")) + .when(engine).refreshData((String) any()); + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setAutomaticUpdatesEnabled(autoUpdateEnabled); + config.setFileSystemWatcherEnabled(false); + + AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); + file.setEngine(engine); + file.setConfiguration(config); + when(engine.getDataFileMetaData()).thenReturn(file); + configureFileUpdate(engine); + + dataUpdate.registerDataFile(file); + int timers = autoUpdateEnabled ? 1 : 0; + ScheduledFuture pending = file.getFuture(); + + // Act + DataUpdateService.AutoUpdateStatus status = + dataUpdate.checkForUpdate(engine); + + // Assert + assertEquals( + DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_REFRESH_FAILED, + status); + verify(futureFactory, times(2 * timers)).schedule( + any(Runnable.class), + anyLong()); + assertEquals( + "The next update timer was not set after the failed refresh.", + autoUpdateEnabled, + file.getFuture() != null); + if (autoUpdateEnabled) { + assertTrue(pending.isCancelled()); + assertNotSame(pending, file.getFuture()); + } + } + + /** + * A data file that has been unregistered is no longer registered, so a + * manual update of it sets no update timer. + */ + @Test + public void DataUpdateService_Unregister_NoLongerRegistered() { + // Arrange + OnPremiseAspectEngine engine = + mock(OnPremiseAspectEngine.class); + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setAutomaticUpdatesEnabled(autoUpdateEnabled); + config.setFileSystemWatcherEnabled(false); + + AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); + file.setEngine(engine); + file.setConfiguration(config); + when(engine.getDataFileMetaData()).thenReturn(file); + configureFileUpdate(engine); + dataUpdate.registerDataFile(file); + int timers = autoUpdateEnabled ? 1 : 0; + + // Act + dataUpdate.unregisterDataFile(file); + DataUpdateService.AutoUpdateStatus status = + dataUpdate.checkForUpdate(engine); + + // Assert + assertFalse(file.getIsRegistered()); + assertEquals(AUTO_UPDATE_SUCCESS, status); + verify(futureFactory, times(timers)).schedule( + any(Runnable.class), + anyLong()); + assertNull(file.getFuture()); + } + + /** + * A manual check that applies an update in memory to a data file + * registered for automatic updates sets the next update timer, as a + * check the timer started does. + */ + @Test + public void DataUpdateService_CheckForUpdate_InMemory_TimerSetAgain() + throws IOException { + // Arrange + configureNoFileSystem(); + when(httpClientConnection.getResponseCode()) + .thenReturn(HttpURLConnection.HTTP_OK); + when(httpClientConnection.getResponseMessage()).thenReturn(null); + when(httpClientConnection.getHeaderField("Content-MD5")) + .thenReturn("08527dcbdd437e7fa6c084423d06dba6"); + when(httpClientConnection.getInputStream()).thenReturn( + DataUpdateServiceTests.class.getClassLoader() + .getResourceAsStream("file.gz")); + + OnPremiseAspectEngine engine = + mock(OnPremiseAspectEngine.class); + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setAutomaticUpdatesEnabled(autoUpdateEnabled); + config.setDataUpdateUrl("http://www.test.com"); + config.setVerifyMd5(true); + config.setDecompressContent(true); + config.setVerifyModifiedSince(false); + + AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); + file.setEngine(engine); + file.setConfiguration(config); + when(engine.getDataFileMetaData()).thenReturn(file); + + dataUpdate.registerDataFile(file); + int timers = autoUpdateEnabled ? 1 : 0; + verify(futureFactory, times(timers)).schedule( + any(Runnable.class), + anyLong()); + + ScheduledFuture pending = file.getFuture(); + + // Act + DataUpdateService.AutoUpdateStatus status = + dataUpdate.checkForUpdate(engine); + + // Assert + assertEquals(AUTO_UPDATE_SUCCESS, status); + verify(futureFactory, times(2 * timers)).schedule( + any(Runnable.class), + anyLong()); + assertEquals( + "The next update timer was not set after the manual update.", + autoUpdateEnabled, + file.getFuture() != null); + + if (autoUpdateEnabled) { + // The pending timer is cancelled, so only the new one is left. + assertTrue( + "The timer pending before the manual update was not cancelled.", + pending.isCancelled()); + assertNotSame(pending, file.getFuture()); + } + } + + /** + * A manual check that finds no update leaves the timer of a data file + * registered for automatic updates as it was. + */ + @Test + public void DataUpdateService_CheckForUpdate_NoUpdate_TimerKept() + throws IOException { + // Arrange + configureNoFileSystem(); + configureHttpNoUpdateAvailable(); + + OnPremiseAspectEngine engine = + mock(OnPremiseAspectEngine.class); + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setAutomaticUpdatesEnabled(autoUpdateEnabled); + config.setDataUpdateUrl("http://www.test.com"); + config.setVerifyModifiedSince(false); + + AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); + file.setEngine(engine); + file.setConfiguration(config); + when(engine.getDataFileMetaData()).thenReturn(file); + + dataUpdate.registerDataFile(file); + Object timer = file.getFuture(); + + // Act + DataUpdateService.AutoUpdateStatus status = + dataUpdate.checkForUpdate(engine); + + // Assert + assertEquals(AUTO_UPDATE_NOT_NEEDED, status); + verify(engine, never()).refreshData(anyString()); + verify(futureFactory, times(autoUpdateEnabled ? 1 : 0)).schedule( + any(Runnable.class), + anyLong()); + assertSame(timer, file.getFuture()); + } + /**