diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java index 8c2169ff7c..40aacb620f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java @@ -183,16 +183,27 @@ public abstract class AbstractRemoteFileStreamingMessageSource @Override public void stop() { if (this.running.compareAndSet(true, false)) { - if (this.filter == null || this.filter.supportsSingleFileFiltering()) { - this.toBeReceived.clear(); - } - else { - // remove unprocessed files from the queue (and filter) - AbstractFileInfo file = this.toBeReceived.poll(); - while (file != null) { - resetFilterIfNecessary(file); - file = this.toBeReceived.poll(); - } + clearFetchedCache(); + } + } + + /** + * Clear internal queue of fetched remote files. + * This functionality might be useful in combination with a + * {@link org.springframework.integration.file.remote.aop.RotatingServerAdvice}, + * when not all fetched files are processed in between rotations. + * @since 6.4 + */ + public void clearFetchedCache() { + if (this.filter == null || this.filter.supportsSingleFileFiltering()) { + this.toBeReceived.clear(); + } + else { + // remove unprocessed files from the queue (and filter) + AbstractFileInfo file = this.toBeReceived.poll(); + while (file != null) { + resetFilterIfNecessary(file); + file = this.toBeReceived.poll(); } } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java index c9c46bf8ab..911eb0d0dc 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2022 the original author or authors. + * Copyright 2015-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,7 +17,6 @@ package org.springframework.integration.file.remote; import java.io.IOException; -import java.io.InputStream; import java.io.UncheckedIOException; import java.util.Collection; import java.util.Comparator; @@ -26,7 +25,6 @@ import java.util.stream.Collectors; import org.junit.jupiter.api.Test; -import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; @@ -37,6 +35,8 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** @@ -49,39 +49,65 @@ import static org.mockito.Mockito.when; public class RemoteFileStreamingMessageSourceTests { @Test - @SuppressWarnings("unchecked") - public void filterOutFilesNotAcceptedByFilter() throws IOException { - RemoteFileTemplate remoteFileTemplate = mock(RemoteFileTemplate.class); + public void fetchFilesFromRemoteAfterClearingFetchedCache() throws IOException { + RemoteFileTemplate remoteFileTemplate = mock(); when(remoteFileTemplate.list("remoteDirectory")).thenReturn(new String[] {"file1", "file2"}); - Session session = mock(Session.class); - when(session.readRaw(anyString())).thenReturn(mock(InputStream.class)); + Session session = mock(); + when(session.readRaw(anyString())).thenReturn(mock()); when(remoteFileTemplate.getSession()).thenReturn(session); - FileListFilter fileListFilter = mock(FileListFilter.class); + Comparator comparator = mock(); + TestRemoteFileStreamingMessageSource testRemoteFileStreamingMessageSource = + new TestRemoteFileStreamingMessageSource(remoteFileTemplate, comparator); + + testRemoteFileStreamingMessageSource.setRemoteDirectory("remoteDirectory"); + testRemoteFileStreamingMessageSource.setBeanFactory(mock()); + testRemoteFileStreamingMessageSource.start(); + + assertThat(testRemoteFileStreamingMessageSource.doReceive(2)) + .isNotNull(); + + testRemoteFileStreamingMessageSource.clearFetchedCache(); + + assertThat(testRemoteFileStreamingMessageSource.doReceive(2)) + .isNotNull(); + + verify(remoteFileTemplate, times(2)).list("remoteDirectory"); + + } + + @Test + public void filterOutFilesNotAcceptedByFilter() throws IOException { + RemoteFileTemplate remoteFileTemplate = mock(); + when(remoteFileTemplate.list("remoteDirectory")).thenReturn(new String[] {"file1", "file2"}); + Session session = mock(); + when(session.readRaw(anyString())).thenReturn(mock()); + when(remoteFileTemplate.getSession()).thenReturn(session); + + FileListFilter fileListFilter = mock(); when(fileListFilter.supportsSingleFileFiltering()).thenReturn(true); when(fileListFilter.accept("file1")).thenReturn(false); when(fileListFilter.accept("file2")).thenReturn(false); - Comparator comparator = mock(Comparator.class); + Comparator comparator = mock(); TestRemoteFileStreamingMessageSource testRemoteFileStreamingMessageSource = new TestRemoteFileStreamingMessageSource(remoteFileTemplate, comparator); testRemoteFileStreamingMessageSource.setFilter(fileListFilter); testRemoteFileStreamingMessageSource.setRemoteDirectory("remoteDirectory"); - testRemoteFileStreamingMessageSource.setBeanFactory(mock(BeanFactory.class)); + testRemoteFileStreamingMessageSource.setBeanFactory(mock()); testRemoteFileStreamingMessageSource.start(); assertThat(testRemoteFileStreamingMessageSource.doReceive(-1)).isNull(); } @Test - @SuppressWarnings("unchecked") public void sessionReturnedToCacheProperlyOnDoReceive() throws IOException { - Session session = mock(Session.class); + Session session = mock(); when(session.readRaw(anyString())).thenThrow(IOException.class); when(session.list("remoteDirectory")).thenReturn(new String[] {"file1"}); - SessionFactory sessionFactory = mock(SessionFactory.class); + SessionFactory sessionFactory = mock(); when(sessionFactory.getSession()).thenReturn(session); CachingSessionFactory cachingSessionFactory = new CachingSessionFactory<>(sessionFactory, 1); @@ -91,7 +117,7 @@ public class RemoteFileStreamingMessageSourceTests { new TestRemoteFileStreamingMessageSource(remoteFileTemplate, null); testRemoteFileStreamingMessageSource.setRemoteDirectory("remoteDirectory"); - testRemoteFileStreamingMessageSource.setBeanFactory(mock(BeanFactory.class)); + testRemoteFileStreamingMessageSource.setBeanFactory(mock()); testRemoteFileStreamingMessageSource.start(); assertThatExceptionOfType(UncheckedIOException.class) diff --git a/src/reference/antora/modules/ROOT/pages/sftp/max-fetch.adoc b/src/reference/antora/modules/ROOT/pages/sftp/max-fetch.adoc index b2c29d2a75..d6e102bd8a 100644 --- a/src/reference/antora/modules/ROOT/pages/sftp/max-fetch.adoc +++ b/src/reference/antora/modules/ROOT/pages/sftp/max-fetch.adoc @@ -25,5 +25,7 @@ If the poller is active when the property is changed, the change takes effect on Starting with version 5.1, the synchronizer can be provided with a `Comparator`. This is useful when restricting the number of files fetched with `maxFetchSize`. -Also see general xref:sftp/inbound.adoc[SFTP Inbound Channel Adapter] chapter for information about `FileListFilter` configuration. +Starting with version 6.4, the `AbstractRemoteFileStreamingMessageSource` has now a convenient `clearFetchedCache()` API to remove references from cache for not processed remote files. +The references stay in cache because polling configuration does not allow to process all of them in one cycle, and the target `SessionFactory` might be changed between polling cycles, e.g. via `RotatingServerAdvice`. +Also see general xref:sftp/inbound.adoc[SFTP Inbound Channel Adapter] chapter for information about `FileListFilter` configuration. diff --git a/src/reference/antora/modules/ROOT/pages/sftp/rotating-server-advice.adoc b/src/reference/antora/modules/ROOT/pages/sftp/rotating-server-advice.adoc index 939165b973..5076432636 100644 --- a/src/reference/antora/modules/ROOT/pages/sftp/rotating-server-advice.adoc +++ b/src/reference/antora/modules/ROOT/pages/sftp/rotating-server-advice.adoc @@ -83,3 +83,4 @@ public IntegrationFlow flow() { IMPORTANT: Do not configure a `TaskExecutor` on the poller when using this advice; see xref:changes-4.1-4.2.adoc#x4.2-conditional-pollers[Conditional Pollers for Message Sources] for more information. +Also see a convenient `AbstractRemoteFileStreamingMessageSource.clearFetchedCache()` API when not all fetched files are processed withing a single polling cycle, but `SessionFactory` might be rotated to different one. \ No newline at end of file diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 6c58a556c8..a08aec7fda 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -17,4 +17,10 @@ In general the project has been moved to the latest dependency versions. === New Components [[x6.4-general]] -=== General Changes \ No newline at end of file +=== General Changes + +[[x6.4-remote-files-changes]] +=== Remote File Adapters Changes + +The `AbstractRemoteFileStreamingMessageSource` has now a convenient `clearFetchedCache()` API to remove references from cache for not processed remote files. +The references stay in cache because polling configuration does not allow to process all the fetched in one cycle, and the target `SessionFactory` might be changed between polling cycles, e.g. via `RotatingServerAdvice`. \ No newline at end of file