diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractFileListFilter.java index f8924a0bbd..bc1d66b806 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractFileListFilter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2019 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. @@ -26,6 +26,7 @@ import java.util.List; * * @author Mark Fisher * @author Iwein Fuld + * @author Gary Russell */ public abstract class AbstractFileListFilter implements FileListFilter { @@ -42,11 +43,17 @@ public abstract class AbstractFileListFilter implements FileListFilter { return accepted; } + @Override + public boolean supportsSingleFileFiltering() { + return true; + } + /** * Subclasses must implement this method. * @param file The file. * @return true if the file passes the filter. */ + @Override public abstract boolean accept(F file); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractMarkerFilePresentFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractMarkerFilePresentFileListFilter.java index c32ec5b729..b1e35acf97 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractMarkerFilePresentFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractMarkerFilePresentFileListFilter.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2019 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. @@ -31,6 +31,9 @@ import java.util.stream.Collectors; * A FileListFilter that only passes files matched by one or more {@link FileListFilter} * if a corresponding marker file is also present to indicate a file transfer is complete. * + * Since they look at multiple files, they cannot be used for late filtering in the + * streaming message source. + * * @author Gary Russell * @since 5.0 * diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java index 3048c817f9..d890e16ef4 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java @@ -173,7 +173,7 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter extends Abst try { this.flushableStore.flush(); } - catch (IOException e) { + catch (@SuppressWarnings("unused") IOException e) { // store's responsibility to log } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/ChainFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/ChainFileListFilter.java index 9764a35fe2..ef95d862e3 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/ChainFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/ChainFileListFilter.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2019 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. @@ -62,4 +62,15 @@ public class ChainFileListFilter extends CompositeFileListFilter { return leftOver; } + @Override + public boolean accept(F file) { + // we can't use stream().allMatch() because there is no guarantee of early exit + for (FileListFilter filter : this.fileFilters) { + if (!filter.accept(file)) { + return false; + } + } + return true; + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java index b96efc355c..c3dce5e3bc 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java @@ -25,6 +25,7 @@ import java.util.Collections; import java.util.LinkedHashSet; import java.util.List; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import org.springframework.beans.factory.InitializingBean; @@ -53,6 +54,8 @@ public class CompositeFileListFilter private Consumer discardCallback; + private boolean allSupportAccept = true; + public CompositeFileListFilter() { this.fileFilters = new LinkedHashSet<>(); @@ -60,6 +63,7 @@ public class CompositeFileListFilter public CompositeFileListFilter(Collection> fileFilters) { this.fileFilters = new LinkedHashSet<>(fileFilters); + this.allSupportAccept = fileFilters.stream().allMatch(FileListFilter::supportsSingleFileFiltering); } @@ -73,6 +77,7 @@ public class CompositeFileListFilter } public CompositeFileListFilter addFilter(FileListFilter filter) { + this.allSupportAccept &= filter.supportsSingleFileFiltering(); return addFilters(Collections.singletonList(filter)); } @@ -81,9 +86,11 @@ public class CompositeFileListFilter * @return this CompositeFileFilter instance with the added filters * @see #addFilters(Collection) */ - @SuppressWarnings("unchecked") - public CompositeFileListFilter addFilters(FileListFilter... filters) { - return addFilters(Arrays.asList(filters)); + @SafeVarargs + @SuppressWarnings("varargs") + public final CompositeFileListFilter addFilters(FileListFilter... filters) { + List> asList = Arrays.asList(filters); + return addFilters(asList); } /** @@ -109,18 +116,19 @@ public class CompositeFileListFilter } } this.fileFilters.addAll(filtersToAdd); + this.allSupportAccept &= filtersToAdd.stream().allMatch(FileListFilter::supportsSingleFileFiltering); return this; } @Override - public void addDiscardCallback(Consumer discardCallback) { - this.discardCallback = discardCallback; + public void addDiscardCallback(Consumer discardCallbackToSet) { + this.discardCallback = discardCallbackToSet; if (this.discardCallback != null) { this.fileFilters .stream() .filter(DiscardAwareFileListFilter.class::isInstance) .map(f -> (DiscardAwareFileListFilter) f) - .forEach(f -> f.addDiscardCallback(discardCallback)); + .forEach(f -> f.addDiscardCallback(discardCallbackToSet)); } } @@ -135,6 +143,19 @@ public class CompositeFileListFilter return results; } + @Override + public boolean accept(F file) { + AtomicBoolean allAccept = new AtomicBoolean(true); + // we can't use stream().allMatch() because we have to call all filters for this filter's contract + this.fileFilters.forEach(f -> allAccept.compareAndSet(true, f.accept(file))); + return allAccept.get(); + } + + @Override + public boolean supportsSingleFileFiltering() { + return this.allSupportAccept; + } + @Override public void rollback(F file, List files) { for (FileListFilter fileFilter : this.fileFilters) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/DiscardAwareFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/DiscardAwareFileListFilter.java index 58deef8ac9..76d5a942b2 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/DiscardAwareFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/DiscardAwareFileListFilter.java @@ -22,9 +22,10 @@ import org.springframework.lang.Nullable; /** * The {@link FileListFilter} modification which can accept a {@link Consumer} - * which can be called when filter discards the file. + * which can be called when the filter discards the file. * * @author Artem Bilan + * @author Gary Russell * * @since 5.0.5 */ diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileListFilter.java index 00518c8bd5..f7eb29981e 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileListFilter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2019 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. @@ -24,6 +24,7 @@ import java.util.List; * * @author Iwein Fuld * @author Josh Long + * @author Gary Russell * * @since 1.0.0 */ @@ -38,4 +39,29 @@ public interface FileListFilter { */ List filterFiles(F[] files); + /** + * Filter a single file; only called externally if {@link #supportsSingleFileFiltering()} + * returns true. + * @param file the file. + * @return true if the file passes the filter, false to filter. + * @since 5.2 + * @see #supportsSingleFileFiltering() + */ + default boolean accept(F file) { + throw new UnsupportedOperationException( + "Filters that return true in supportsSingleFileFiltering() must implement this method"); + } + + /** + * Indicates that this filter supports filtering a single file. + * Filters that return true must override {@link #accept(Object)}. + * Default false. + * @return true to allow external calls to {@link #accept(Object)}. + * @since 5.2 + * @see #accept(Object) + */ + default boolean supportsSingleFileFiltering() { + return false; + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/LastModifiedFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/LastModifiedFileListFilter.java index 111493562c..856cb5488a 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/LastModifiedFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/LastModifiedFileListFilter.java @@ -105,8 +105,8 @@ public class LastModifiedFileListFilter implements DiscardAwareFileListFilter discardCallback) { - this.discardCallback = discardCallback; + public void addDiscardCallback(@Nullable Consumer discardCallbackToSet) { + this.discardCallback = discardCallbackToSet; } @Override @@ -114,7 +114,7 @@ public class LastModifiedFileListFilter implements DiscardAwareFileListFilter list = new ArrayList<>(); long now = System.currentTimeMillis() / ONE_SECOND; for (File file : files) { - if (file.lastModified() / ONE_SECOND + this.age <= now) { + if (fileIsAged(file, now)) { list.add(file); } else if (this.discardCallback != null) { @@ -124,4 +124,25 @@ public class LastModifiedFileListFilter implements DiscardAwareFileListFilter /** * Subclasses can override to perform initialization - called from - * {@link InitializingBean#afterPropertiesSet()}. + * {@link org.springframework.beans.factory.InitializingBean#afterPropertiesSet()}. */ protected void doInit() { } @@ -162,11 +161,16 @@ public abstract class AbstractRemoteFileStreamingMessageSource @Override public void stop() { if (this.running.compareAndSet(true, false)) { - // remove unprocessed files from the queue (and filter) - AbstractFileInfo file = this.toBeReceived.poll(); - while (file != null) { - resetFilterIfNecessary(file); - file = this.toBeReceived.poll(); + 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(); + } } } } @@ -180,26 +184,38 @@ public abstract class AbstractRemoteFileStreamingMessageSource protected Object doReceive() { Assert.state(this.running.get(), () -> getComponentName() + " is not running"); AbstractFileInfo file = poll(); - if (file != null) { - try { - String remotePath = remotePath(file); - Session session = this.remoteFileTemplate.getSession(); - try { - return getMessageBuilderFactory() - .withPayload(session.readRaw(remotePath)) - .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session) - .setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory()) - .setHeader(FileHeaders.REMOTE_FILE, file.getFilename()) - .setHeader(FileHeaders.REMOTE_FILE_INFO, - this.fileInfoJson ? file.toJson() : file); - } - catch (IOException e) { - throw new UncheckedIOException("IOException when retrieving " + remotePath, e); - } + while (file != null) { + if (this.filter != null && this.filter.supportsSingleFileFiltering() + && !this.filter.accept(file.getFileInfo())) { + + if (this.toBeReceived.size() > 0) { // don't re-fetch already filtered files + file = poll(); + } + else { + file = null; + } } - catch (RuntimeException e) { - resetFilterIfNecessary(file); - throw e; + if (file != null) { + try { + String remotePath = remotePath(file); + Session session = this.remoteFileTemplate.getSession(); + try { + return getMessageBuilderFactory() + .withPayload(session.readRaw(remotePath)) + .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session) + .setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory()) + .setHeader(FileHeaders.REMOTE_FILE, file.getFilename()) + .setHeader(FileHeaders.REMOTE_FILE_INFO, + this.fileInfoJson ? file.toJson() : file); + } + catch (IOException e) { + throw new UncheckedIOException("IOException when retrieving " + remotePath, e); + } + } + catch (RuntimeException e) { + resetFilterIfNecessary(file); + throw e; + } } } return null; @@ -240,17 +256,23 @@ public abstract class AbstractRemoteFileStreamingMessageSource files = FileUtils.purgeUnwantedElements(files, f -> f == null || isDirectory(f), this.comparator); } if (!ObjectUtils.isEmpty(files)) { - int maxFetchSize = getMaxFetchSize(); - List filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files); - if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) { - rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); - List newList = new ArrayList<>(maxFetchSize); - for (int i = 0; i < maxFetchSize; i++) { - newList.add(filteredFiles.get(i)); + List> fileInfoList; + if (this.filter != null && !this.filter.supportsSingleFileFiltering()) { + int maxFetchSize = getMaxFetchSize(); + List filteredFiles = this.filter.filterFiles(files); + if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) { + rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); + List newList = new ArrayList<>(maxFetchSize); + for (int i = 0; i < maxFetchSize; i++) { + newList.add(filteredFiles.get(i)); + } + filteredFiles = newList; } - filteredFiles = newList; + fileInfoList = asFileInfoList(filteredFiles); + } + else { + fileInfoList = asFileInfoList(Arrays.asList(files)); } - List> fileInfoList = asFileInfoList(filteredFiles); fileInfoList.forEach(fi -> fi.setRemoteDirectory(remoteDirectory)); this.toBeReceived.addAll(fileInfoList); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java index f2f2d235c4..4ed6b2f6a0 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java @@ -22,11 +22,11 @@ import java.io.File; import java.io.FileOutputStream; import java.io.IOException; import java.io.OutputStream; -import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; import java.util.List; import java.util.regex.Matcher; +import java.util.stream.Collectors; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -215,9 +215,9 @@ public abstract class AbstractInboundFileSynchronizer } - protected final void doSetRemoteDirectoryExpression(Expression remoteDirectoryExpression) { - Assert.notNull(remoteDirectoryExpression, "'remoteDirectoryExpression' must not be null"); - this.remoteDirectoryExpression = remoteDirectoryExpression; + protected final void doSetRemoteDirectoryExpression(Expression expression) { + Assert.notNull(expression, "'remoteDirectoryExpression' must not be null"); + this.remoteDirectoryExpression = expression; evaluateRemoteDirectory(); } @@ -229,8 +229,8 @@ public abstract class AbstractInboundFileSynchronizer doSetFilter(filter); } - protected final void doSetFilter(@Nullable FileListFilter filter) { - this.filter = filter; + protected final void doSetFilter(@Nullable FileListFilter filterToSet) { + this.filter = filterToSet; } /** @@ -325,19 +325,23 @@ public abstract class AbstractInboundFileSynchronizer files = FileUtils.purgeUnwantedElements(files, e -> !isFile(e), this.comparator); } if (!ObjectUtils.isEmpty(files)) { - List filteredFiles = filterFiles(files); - if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) { - rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); - List newList = new ArrayList<>(maxFetchSize); - for (int i = 0; i < maxFetchSize; i++) { - newList.add(filteredFiles.get(i)); - } - filteredFiles = newList; - } + boolean haveFilter = this.filter != null; + boolean filteringOneByOne = haveFilter && this.filter.supportsSingleFileFiltering(); + List filteredFiles = applyFilter(files, haveFilter, filteringOneByOne, maxFetchSize); int copied = filteredFiles.size(); + int accepted = 0; for (F file : filteredFiles) { + if (filteringOneByOne) { + if ((maxFetchSize < 0 || accepted < maxFetchSize) && this.filter.accept(file)) { + accepted++; + } + else { + file = null; + copied--; + } + } try { if (file != null && !copyFileToLocalDirectory(this.evaluatedRemoteDirectory, file, localDirectory, session)) { @@ -345,7 +349,12 @@ public abstract class AbstractInboundFileSynchronizer } } catch (RuntimeException | IOException e1) { - rollbackFromFileToListEnd(filteredFiles, file); + if (filteringOneByOne) { + resetFilterIfNecessary(file); + } + else { + rollbackFromFileToListEnd(filteredFiles, file); + } throw e1; } } @@ -356,6 +365,27 @@ public abstract class AbstractInboundFileSynchronizer } } + private List applyFilter(F[] files, boolean haveFilter, boolean filteringOneByOne, int maxFetchSize) { + List filteredFiles; + if (!filteringOneByOne && haveFilter) { + filteredFiles = filterFiles(files); + } + else { + filteredFiles = Arrays.asList(files); + } + if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) { + if (!filteringOneByOne) { + if (haveFilter) { + rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); + } + filteredFiles = filteredFiles.stream() + .limit(maxFetchSize) + .collect(Collectors.toList()); + } + } + return filteredFiles; + } + protected void rollbackFromFileToListEnd(List filteredFiles, F file) { if (this.filter instanceof ReversibleFileListFilter) { ((ReversibleFileListFilter) this.filter) @@ -418,12 +448,8 @@ public abstract class AbstractInboundFileSynchronizer } return true; } - else if (this.filter instanceof ResettableFileListFilter) { - if (this.logger.isInfoEnabled()) { - this.logger.info("Reverting the remote file '" + remoteFile + - "' from the filter for a subsequent transfer attempt"); - } - ((ResettableFileListFilter) this.filter).remove(remoteFile); + else { + resetFilterIfNecessary(remoteFile); } } else if (this.logger.isWarnEnabled()) { @@ -434,6 +460,16 @@ public abstract class AbstractInboundFileSynchronizer return false; } + private void resetFilterIfNecessary(F remoteFile) { + if (this.filter instanceof ResettableFileListFilter) { + if (this.logger.isInfoEnabled()) { + this.logger.info("Removing the remote file '" + remoteFile + + "' from the filter for a subsequent transfer attempt"); + } + ((ResettableFileListFilter) this.filter).remove(remoteFile); + } + } + private boolean copyRemoteContentToLocalFile(Session session, String remoteFilePath, File localFile) { boolean renamed; String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix; diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/CompositeFileListFilterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/CompositeFileListFilterTests.java index 7e61be2119..2e1a68bc7d 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/CompositeFileListFilterTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/CompositeFileListFilterTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.file.filters; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.ArgumentMatchers.isA; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -24,6 +25,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.File; +import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -101,4 +103,59 @@ public class CompositeFileListFilterTests { compositeFileFilter.close(); } + @Test + public void singleFileCapableUO() throws IOException { + CompositeFileListFilter compo = + new CompositeFileListFilter<>(Collections.singletonList(new FileListFilter() { + + @Override + public List filterFiles(String[] files) { + return Collections.emptyList(); + } + + @Override + public boolean supportsSingleFileFiltering() { + return true; + } + + })); + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(() -> compo.accept("foo")); + compo.close(); + } + + @Test + public void singleFileCapable() throws IOException { + CompositeFileListFilter compo = + new CompositeFileListFilter<>(Collections.singletonList(new FileListFilter() { + + @Override + public List filterFiles(String[] files) { + return Collections.emptyList(); + } + + @Override + public boolean supportsSingleFileFiltering() { + return true; + } + + @Override + public boolean accept(String file) { + return true; + } + + })); + assertThat(compo.accept("foo")).isTrue(); + compo.addFilter(s -> null); + assertThat(compo.supportsSingleFileFiltering()).isFalse(); + compo.close(); + } + + @Test + public void notSingleFileCapable() throws IOException { + CompositeFileListFilter compo = + new CompositeFileListFilter<>(Collections.singletonList(s -> null)); + assertThat(compo.supportsSingleFileFiltering()).isFalse(); + compo.close(); + } + } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/LastModifiedFileListFilterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/LastModifiedFileListFilterTests.java index e91f99f568..bc5050d4c5 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/LastModifiedFileListFilterTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/LastModifiedFileListFilterTests.java @@ -45,10 +45,12 @@ public class LastModifiedFileListFilterTests { FileOutputStream fileOutputStream = new FileOutputStream(foo); fileOutputStream.write("x".getBytes()); fileOutputStream.close(); - assertThat(filter.filterFiles(new File[] { foo }).size()).isEqualTo(0); + assertThat(filter.filterFiles(new File[] { foo })).hasSize(0); + assertThat(filter.accept(foo)).isFalse(); // Make a file as of yesterday's foo.setLastModified(System.currentTimeMillis() - 1000 * 60 * 60 * 24); - assertThat(filter.filterFiles(new File[] { foo }).size()).isEqualTo(1); + assertThat(filter.filterFiles(new File[] { foo })).hasSize(1); + assertThat(filter.accept(foo)).isTrue(); } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java index 7d5826d886..eeb43970f0 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java @@ -35,6 +35,8 @@ import java.util.Comparator; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; +import java.util.stream.Stream; import org.junit.Test; @@ -45,6 +47,7 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; +import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.file.splitter.FileSplitter; @@ -65,13 +68,33 @@ public class StreamingInboundTests { private final StreamTransformer transformer = new StreamTransformer(); - @SuppressWarnings("unchecked") @Test - public void testAllData() throws Exception { + public void testAllDataNoFilter() throws IOException { + testAllData(null, true); + } + + @Test + public void testAllDataSingleCapableFilter() throws IOException { + testAllData(null, false); + } + + @Test + public void testAllDataBulkOnlyFilter() throws IOException { + testAllData(fs -> Stream.of(fs).collect(Collectors.toList()), false); + } + + @SuppressWarnings("unchecked") + private void testAllData(FileListFilter filter, boolean nullFilter) throws IOException { StringSessionFactory sessionFactory = new StringSessionFactory(); Streamer streamer = new Streamer(new StringRemoteFileTemplate(sessionFactory), null); streamer.setBeanFactory(mock(BeanFactory.class)); streamer.setRemoteDirectory("/foo"); + if (filter != null) { + streamer.setFilter(filter); + } + if (nullFilter) { + streamer.setFilter(null); + } streamer.afterPropertiesSet(); streamer.start(); Message received = (Message) this.transformer.transform(streamer.receive()); @@ -116,7 +139,6 @@ public class StreamingInboundTests { Streamer streamer = new Streamer(new StringRemoteFileTemplate(sessionFactory), null); streamer.setBeanFactory(mock(BeanFactory.class)); streamer.setRemoteDirectory("/foo"); - streamer.setMaxFetchSize(1); streamer.setFilter(new AcceptOnceFileListFilter<>()); streamer.afterPropertiesSet(); streamer.start(); @@ -133,10 +155,10 @@ public class StreamingInboundTests { assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("bar"); - // close after list, transform - verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(4)).close(); + // close after transform + verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(3)).close(); - verify(sessionFactory.getSession(), times(2)).list("/foo"); + verify(sessionFactory.getSession()).list("/foo"); } @Test @@ -204,10 +226,9 @@ public class StreamingInboundTests { streamer.start(); assertThat(streamer.receive()).isNotNull(); assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(1); - assertThat(streamer.metadataMap).hasSize(2); + assertThat(streamer.metadataMap).hasSize(1); streamer.stop(); assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(0); - assertThat(streamer.metadataMap).hasSize(1); } @SuppressWarnings("unchecked") @@ -221,13 +242,18 @@ public class StreamingInboundTests { assertThatExceptionOfType(UncheckedIOException.class) .isThrownBy(streamer::receive); assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(1); - assertThat(streamer.metadataMap).hasSize(1); + assertThat(streamer.metadataMap).hasSize(0); } public static class Streamer extends AbstractRemoteFileStreamingMessageSource { ConcurrentHashMap metadataMap = new ConcurrentHashMap<>(); + protected Streamer(RemoteFileTemplate template) { + super(template, null); + doSetFilter(null); + } + protected Streamer(RemoteFileTemplate template, Comparator comparator) { super(template, comparator); doSetFilter(new StringPersistentFileListFilter(new SimpleMetadataStore(this.metadataMap), "streamer")); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java index 1b30742544..6aa038a977 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java @@ -27,12 +27,15 @@ import java.io.OutputStream; import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import java.util.stream.Stream; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.file.HeadDirectoryScanner; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; +import org.springframework.integration.file.filters.ChainFileListFilter; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.messaging.MessagingException; @@ -113,7 +116,7 @@ public class AbstractRemoteFileSynchronizerTests { } @Test - public void testMaxFetchSizeSource() throws Exception { + public void testMaxFetchSizeSource() { final AtomicInteger count = new AtomicInteger(); AbstractInboundFileSynchronizer sync = createLimitingSynchronizer(count); AbstractInboundFileSynchronizingMessageSource source = createSource(sync); @@ -130,7 +133,62 @@ public class AbstractRemoteFileSynchronizerTests { } @Test - public void testExclusiveScanner() throws Exception { + public void testDefaultFilter() { + final AtomicInteger count = new AtomicInteger(); + AbstractInboundFileSynchronizingMessageSource source = createSource(count); + source.afterPropertiesSet(); + source.start(); + source.receive(); + assertThat(count.get()).isEqualTo(1); + source.receive(); + assertThat(count.get()).isEqualTo(2); + source.receive(); + assertThat(count.get()).isEqualTo(3); + source.receive(); + assertThat(count.get()).isEqualTo(3); + } + + @Test + public void testNoFilter() { + final AtomicInteger count = new AtomicInteger(); + AbstractInboundFileSynchronizer sync = createLimitingSynchronizer(count); + sync.setFilter(null); + AbstractInboundFileSynchronizingMessageSource source = createSource(sync); + source.afterPropertiesSet(); + source.start(); + source.receive(); + assertThat(count.get()).isEqualTo(1); + source.receive(); + assertThat(count.get()).isEqualTo(2); + source.receive(); + assertThat(count.get()).isEqualTo(3); + source.receive(); + assertThat(count.get()).isEqualTo(4); + } + + @Test + public void testBulkOnlyFilter() { + final AtomicInteger count = new AtomicInteger(); + AbstractInboundFileSynchronizer sync = createLimitingSynchronizer(count); + ChainFileListFilter cflf = new ChainFileListFilter<>(); + cflf.addFilter(new AcceptOnceFileListFilter<>()); + cflf.addFilter(fs -> Stream.of(fs).collect(Collectors.toList())); + sync.setFilter(cflf); + AbstractInboundFileSynchronizingMessageSource source = createSource(sync); + source.afterPropertiesSet(); + source.start(); + source.receive(); + assertThat(count.get()).isEqualTo(1); + source.receive(); + assertThat(count.get()).isEqualTo(2); + source.receive(); + assertThat(count.get()).isEqualTo(3); + source.receive(); + assertThat(count.get()).isEqualTo(3); + } + + @Test + public void testExclusiveScanner() { final AtomicInteger count = new AtomicInteger(); AbstractInboundFileSynchronizingMessageSource source = createSource(count); source.setScanner(new HeadDirectoryScanner(1)); @@ -141,7 +199,7 @@ public class AbstractRemoteFileSynchronizerTests { } @Test - public void testExclusiveWatchService() throws Exception { + public void testExclusiveWatchService() { final AtomicInteger count = new AtomicInteger(); AbstractInboundFileSynchronizingMessageSource source = createSource(count); source.setUseWatchService(true); @@ -152,7 +210,7 @@ public class AbstractRemoteFileSynchronizerTests { } @Test(expected = IllegalStateException.class) - public void testScannerAndWatchServiceConflict() throws Exception { + public void testScannerAndWatchServiceConflict() { final AtomicInteger count = new AtomicInteger(); AbstractInboundFileSynchronizingMessageSource source = createSource(count); source.setUseWatchService(true); @@ -166,6 +224,7 @@ public class AbstractRemoteFileSynchronizerTests { private AbstractInboundFileSynchronizingMessageSource createSource( AbstractInboundFileSynchronizer sync) { + AbstractInboundFileSynchronizingMessageSource source = new AbstractInboundFileSynchronizingMessageSource(sync) { @@ -204,7 +263,7 @@ public class AbstractRemoteFileSynchronizerTests { @Override protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, - File localDirectory, Session session) throws IOException { + File localDirectory, Session session) { count.incrementAndGet(); return true; } @@ -227,40 +286,44 @@ public class AbstractRemoteFileSynchronizerTests { private class StringSession implements Session { + StringSession() { + super(); + } + @Override - public boolean remove(String path) throws IOException { + public boolean remove(String path) { return true; } @Override - public String[] list(String path) throws IOException { + public String[] list(String path) { return new String[] { "foo", "bar", "baz" }; } @Override - public void read(String source, OutputStream outputStream) throws IOException { + public void read(String source, OutputStream outputStream) { } @Override - public void write(InputStream inputStream, String destination) throws IOException { + public void write(InputStream inputStream, String destination) { } @Override - public void append(InputStream inputStream, String destination) throws IOException { + public void append(InputStream inputStream, String destination) { } @Override - public boolean mkdir(String directory) throws IOException { + public boolean mkdir(String directory) { return true; } @Override - public boolean rmdir(String directory) throws IOException { + public boolean rmdir(String directory) { return true; } @Override - public void rename(String pathFrom, String pathTo) throws IOException { + public void rename(String pathFrom, String pathTo) { } @Override @@ -273,22 +336,22 @@ public class AbstractRemoteFileSynchronizerTests { } @Override - public boolean exists(String path) throws IOException { + public boolean exists(String path) { return true; } @Override - public String[] listNames(String path) throws IOException { + public String[] listNames(String path) { return new String[0]; } @Override - public InputStream readRaw(String source) throws IOException { + public InputStream readRaw(String source) { return null; } @Override - public boolean finalizeRaw() throws IOException { + public boolean finalizeRaw() { return true; } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java index c4b76b8e1c..a0169355e2 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java @@ -124,10 +124,9 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { assertThat(received).isNotNull(); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO)).isInstanceOf(FtpFileInfo.class); assertThat(TestUtils.getPropertyValue(source, "toBeReceived", BlockingQueue.class)).hasSize(1); - assertThat(this.metadataMap).hasSize(2); + assertThat(this.metadataMap).hasSize(1); this.adapter.stop(); assertThat(TestUtils.getPropertyValue(source, "toBeReceived", BlockingQueue.class)).isEmpty(); - assertThat(this.metadataMap).hasSize(1); } @Test @@ -166,7 +165,6 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(this.config.template(), Comparator.comparing(FTPFile::getName)); messageSource.setRemoteDirectory("ftpSource/"); - messageSource.setMaxFetchSize(1); messageSource.setBeanFactory(this.context); return messageSource; } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java index c52a121605..ad0e5469d2 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java index e9bdca6009..80609e260e 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java @@ -169,7 +169,6 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { new SftpStreamingMessageSource(this.config.template(), Comparator.comparing(LsEntry::getFilename)); messageSource.setRemoteDirectory("sftpSource/"); - messageSource.setMaxFetchSize(1); messageSource.setBeanFactory(this.context); return messageSource; } diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 393601f884..3d68f9a2f2 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -461,7 +461,7 @@ public class FileReadingJavaApplication { .transform(Files.toStringTransformer()) .channel("processFileChannel") .get(); - } + } } ---- @@ -482,15 +482,15 @@ Examples of such events include the following: ==== [source,bash] ---- -[message=tail: cannot open `/tmp/somefile' for reading: +[message=tail: cannot open '/tmp/somefile' for reading: No such file or directory, file=/tmp/somefile] -[message=tail: `/tmp/somefile' has become accessible, file=/tmp/somefile] +[message=tail: '/tmp/somefile' has become accessible, file=/tmp/somefile] -[message=tail: `/tmp/somefile' has become inaccessible: +[message=tail: '/tmp/somefile' has become inaccessible: No such file or directory, file=/tmp/somefile] -[message=tail: `/tmp/somefile' has appeared; +[message=tail: '/tmp/somefile' has appeared; following end of new file, file=/tmp/somefile] ---- ==== @@ -1114,3 +1114,33 @@ public class FileSplitterApplication { } ---- ==== + +[[remote-persistent-flf]] +=== Remote Persistent File List Filters + +Inbound and streaming inbound remote file channel adapters (`FTP`, `SFTP`, and other technologies) are configured with corresponding implementations of `AbstractPersistentFileListFilter` by default, configured with an in-memory `MetadataStore`. +To run in a cluster, these can be replaced with filters using a shared `MetadataStore` (see <> for more information). +These filters are used to prevent fetching the same file multiple times (unless it's modified time changes). +Starting with version 5.2, a file is added to the filter immediately before the file is fetched (and reversed if the fetch fails). + +IMPORTANT: In the event of a catastrophic failure (such as power loss), it is possible that the file currently being fetched will remain in the filter and won't be re-fetched when restarting the application. +In this case you would need to manually remove this file from the `MetadataStore`. + +In previous versions, the files were filtered before any were fetched, meaning that several files could be in this state after a catastrophic failure. + +In order to facilitate this new behavior, two new methods have been added to `FileListFilter`. + +==== +[source, java] +---- +boolean accept(F file); + +boolean supportsSingleFileFiltering(); +---- +==== + +If a filter returns `true` in `supportsSingleFileFiltering`, it **must** implement `accept()`. + +If a remote filter does not support single file filtering (such as the `AbstractMarkerFilePresentFileListFilter`), the adapters revert to the previous behavior. + +If multiple filters are in used (using a `CompositeFileListFilter` or `ChainFileListFilter`), then **all** of the delegate filters must support single file filtering in order for the composite filter to support it. diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 40b181b694..c8f000a82d 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -344,6 +344,8 @@ Unless your application removes files after processing, the adapter will re-proc Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter` and the remote file timestamp changes (causing it to be re-fetched), the default local filter does not let this new file be processed. +For more information about this filter, and how it is used, see <>. + You can use the `local-filter` attribute to configure the behavior of the local file system filter. Starting with version 4.3.8, a `FileSystemPersistentAcceptOnceFileListFilter` is configured by default. This filter stores the accepted file names and modified timestamp in an instance of the `MetadataStore` strategy (see <>) and detects changes to the local file modified time. @@ -623,6 +625,8 @@ If you need to allow duplicates, you can use `AcceptAllFileListFilter`. Any other use cases can be handled by `CompositeFileListFilter` (or `ChainFileListFilter`). The Java configuration (<>) shows one technique to remove the remote file after processing to avoid duplicates. +For more information about the `FtpPersistentAcceptOnceFileListFilter`, and how it is used, see <>. + Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary. Set it to `1` and use a persistent filter when running in a clustered environment. See <> for more information. diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index ef77665adf..0a6147f196 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -371,7 +371,9 @@ Once the files have been retrieved, an additional filter is applied to the files By default, this is an`AcceptOnceFileListFilter`, which, as discussed in this section, retains state in memory and does not consider the file's modified time. Unless your application removes files after processing, the adapter re-processes the files on disk by default after an application restart. -Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter` and the remote file timestamp changes (causing it to be re-fetched), the default local filter does not allow this new file to be processed. +Also, if you configure the `filter` to use a `SftpPersistentAcceptOnceFileListFilter` and the remote file timestamp changes (causing it to be re-fetched), the default local filter does not allow this new file to be processed. + +For more information about this filter, and how it is used, see <>. You can use the `local-filter` attribute to configure the behavior of the local file system filter. Starting with version 4.3.8, a `FileSystemPersistentAcceptOnceFileListFilter` is configured by default. @@ -622,6 +624,8 @@ If you need to allow duplicates, you can use the `AcceptAllFileListFilter`. You can handle any other use cases by using `CompositeFileListFilter` (or `ChainFileListFilter`). The Java configuration <> shows one technique to remove the remote file after processing, avoiding duplicates. +For more information about the `SftpPersistentAcceptOnceFileListFilter`, and how it is used, see <>. + You can use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary. Set it to `1` and use a persistent filter when running in a clustered environment. See <> for more information. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 22d9b701c6..645a4b65e9 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -10,7 +10,13 @@ If you are interested in more details, see the Issue Tracker tickets that were r [[x5.2-general]] === General Changes -[[5.2-tcp]] +[[x5.2-file]] +==== File Changes + +Some improvements to filtering remote files have been made. +See <> for more information. + +[[x5.2-tcp]] ==== TCP Changes The length header used by the `ByteArrayLengthHeaderSerializer` can now include the length of the header in addition to the payload.