From 931df86274c116f806b059f98a375b38dbd0fd77 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 1 Mar 2019 17:29:32 -0500 Subject: [PATCH] GH-2777: Remote File Filter Improvements Resolves https://github.com/spring-projects/spring-integration/issues/2777 If the filter supports it, defer filtering until the last possible moment. Then, the worst case scenario after a catastrophic failure (e.g. power loss), would be that at most one file will be incorrectly filtered on restart. Polishing and add more tests. Polishing Javadocs More Polishing Final polishing More polishing. Polishing and docPolishing and docs. * Fix typos in Docs --- .../file/filters/AbstractFileListFilter.java | 9 +- ...stractMarkerFilePresentFileListFilter.java | 5 +- ...actPersistentAcceptOnceFileListFilter.java | 2 +- .../file/filters/ChainFileListFilter.java | 13 ++- .../file/filters/CompositeFileListFilter.java | 33 +++++-- .../filters/DiscardAwareFileListFilter.java | 3 +- .../file/filters/FileListFilter.java | 28 +++++- .../filters/LastModifiedFileListFilter.java | 27 +++++- ...tractRemoteFileStreamingMessageSource.java | 92 +++++++++++------- .../AbstractInboundFileSynchronizer.java | 80 ++++++++++----- .../filters/CompositeFileListFilterTests.java | 57 +++++++++++ .../LastModifiedFileListFilterTests.java | 6 +- .../file/remote/StreamingInboundTests.java | 44 +++++++-- .../AbstractRemoteFileSynchronizerTests.java | 97 +++++++++++++++---- .../FtpStreamingMessageSourceTests.java | 4 +- .../ftp/inbound/RotatingServersTests.java | 2 +- .../SftpStreamingMessageSourceTests.java | 1 - src/reference/asciidoc/file.adoc | 40 +++++++- src/reference/asciidoc/ftp.adoc | 4 + src/reference/asciidoc/sftp.adoc | 6 +- src/reference/asciidoc/whats-new.adoc | 8 +- 21 files changed, 449 insertions(+), 112 deletions(-) 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.