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

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<F> implements FileListFilter<F> {
 
@@ -42,11 +43,17 @@ public abstract class AbstractFileListFilter<F> implements FileListFilter<F> {
 		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);
 
 }

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
  *

spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java

@@ -173,7 +173,7 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 			try {
 				this.flushableStore.flush();
 			}
-			catch (IOException e) {
+			catch (@SuppressWarnings("unused") IOException e) {
 				// store's responsibility to log
 			}
 		}

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<F> extends CompositeFileListFilter<F> {
 		return leftOver;
 	}
 
+	@Override
+	public boolean accept(F file) {
+		// we can't use stream().allMatch() because there is no guarantee of early exit
+		for (FileListFilter<F> filter : this.fileFilters) {
+			if (!filter.accept(file)) {
+				return false;
+			}
+		}
+		return true;
+	}
+
 }

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<F>
 
 	private Consumer<F> discardCallback;
 
+	private boolean allSupportAccept = true;
+
 
 	public CompositeFileListFilter() {
 		this.fileFilters = new LinkedHashSet<>();
@@ -60,6 +63,7 @@ public class CompositeFileListFilter<F>
 
 	public CompositeFileListFilter(Collection<? extends FileListFilter<F>> fileFilters) {
 		this.fileFilters = new LinkedHashSet<>(fileFilters);
+		this.allSupportAccept = fileFilters.stream().allMatch(FileListFilter<F>::supportsSingleFileFiltering);
 	}
 
 
@@ -73,6 +77,7 @@ public class CompositeFileListFilter<F>
 	}
 
 	public CompositeFileListFilter<F> addFilter(FileListFilter<F> filter) {
+		this.allSupportAccept &= filter.supportsSingleFileFiltering();
 		return addFilters(Collections.singletonList(filter));
 	}
 
@@ -81,9 +86,11 @@ public class CompositeFileListFilter<F>
 	 * @return this CompositeFileFilter instance with the added filters
 	 * @see #addFilters(Collection)
 	 */
-	@SuppressWarnings("unchecked")
-	public CompositeFileListFilter<F> addFilters(FileListFilter<F>... filters) {
-		return addFilters(Arrays.asList(filters));
+	@SafeVarargs
+	@SuppressWarnings("varargs")
+	public final CompositeFileListFilter<F> addFilters(FileListFilter<F>... filters) {
+		List<FileListFilter<F>> asList = Arrays.asList(filters);
+		return addFilters(asList);
 	}
 
 	/**
@@ -109,18 +116,19 @@ public class CompositeFileListFilter<F>
 			}
 		}
 		this.fileFilters.addAll(filtersToAdd);
+		this.allSupportAccept &= filtersToAdd.stream().allMatch(FileListFilter<F>::supportsSingleFileFiltering);
 		return this;
 	}
 
 	@Override
-	public void addDiscardCallback(Consumer<F> discardCallback) {
-		this.discardCallback = discardCallback;
+	public void addDiscardCallback(Consumer<F> discardCallbackToSet) {
+		this.discardCallback = discardCallbackToSet;
 		if (this.discardCallback != null) {
 			this.fileFilters
 					.stream()
 					.filter(DiscardAwareFileListFilter.class::isInstance)
 					.map(f -> (DiscardAwareFileListFilter<F>) f)
-					.forEach(f -> f.addDiscardCallback(discardCallback));
+					.forEach(f -> f.addDiscardCallback(discardCallbackToSet));
 		}
 	}
 
@@ -135,6 +143,19 @@ public class CompositeFileListFilter<F>
 		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<F> files) {
 		for (FileListFilter<F> fileFilter : this.fileFilters) {

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

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<F> {
 	 */
 	List<F> 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 <b>must</b> 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;
+	}
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/filters/LastModifiedFileListFilter.java

@@ -105,8 +105,8 @@ public class LastModifiedFileListFilter implements DiscardAwareFileListFilter<Fi
 	}
 
 	@Override
-	public void addDiscardCallback(@Nullable Consumer<File> discardCallback) {
-		this.discardCallback = discardCallback;
+	public void addDiscardCallback(@Nullable Consumer<File> discardCallbackToSet) {
+		this.discardCallback = discardCallbackToSet;
 	}
 
 	@Override
@@ -114,7 +114,7 @@ public class LastModifiedFileListFilter implements DiscardAwareFileListFilter<Fi
 		List<File> 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<Fi
 		return list;
 	}
 
+	@Override
+	public boolean accept(File file) {
+		if (fileIsAged(file, System.currentTimeMillis() / ONE_SECOND)) {
+			return true;
+		}
+		else if (this.discardCallback != null) {
+			this.discardCallback.accept(file);
+		}
+		return false;
+	}
+
+	private boolean fileIsAged(File file, long now) {
+		return file.lastModified() / ONE_SECOND + this.age <= now;
+	}
+
+	@Override
+	public boolean supportsSingleFileFiltering() {
+		return true;
+	}
+
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java

@@ -28,7 +28,6 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.Lifecycle;
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
@@ -148,7 +147,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
 
 	/**
 	 * 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<F>
 	@Override
 	public void stop() {
 		if (this.running.compareAndSet(true, false)) {
-			// remove unprocessed files from the queue (and filter)
-			AbstractFileInfo<F> 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<F> file = this.toBeReceived.poll();
+				while (file != null) {
+					resetFilterIfNecessary(file);
+					file = this.toBeReceived.poll();
+				}
 			}
 		}
 	}
@@ -180,26 +184,38 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
 	protected Object doReceive() {
 		Assert.state(this.running.get(), () -> getComponentName() + " is not running");
 		AbstractFileInfo<F> 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<F>
 			files = FileUtils.purgeUnwantedElements(files, f -> f == null || isDirectory(f), this.comparator);
 		}
 		if (!ObjectUtils.isEmpty(files)) {
-			int maxFetchSize = getMaxFetchSize();
-			List<F> filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files);
-			if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) {
-				rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));
-				List<F> newList = new ArrayList<>(maxFetchSize);
-				for (int i = 0; i < maxFetchSize; i++) {
-					newList.add(filteredFiles.get(i));
+			List<AbstractFileInfo<F>> fileInfoList;
+			if (this.filter != null && !this.filter.supportsSingleFileFiltering()) {
+				int maxFetchSize = getMaxFetchSize();
+				List<F> filteredFiles = this.filter.filterFiles(files);
+				if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) {
+					rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));
+					List<F> 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<AbstractFileInfo<F>> fileInfoList = asFileInfoList(filteredFiles);
 			fileInfoList.forEach(fi -> fi.setRemoteDirectory(remoteDirectory));
 			this.toBeReceived.addAll(fileInfoList);
 		}

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<F>
 	}
 
 
-	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<F>
 		doSetFilter(filter);
 	}
 
-	protected final void doSetFilter(@Nullable FileListFilter<F> filter) {
-		this.filter = filter;
+	protected final void doSetFilter(@Nullable FileListFilter<F> filterToSet) {
+		this.filter = filterToSet;
 	}
 
 	/**
@@ -325,19 +325,23 @@ public abstract class AbstractInboundFileSynchronizer<F>
 			files = FileUtils.purgeUnwantedElements(files, e -> !isFile(e), this.comparator);
 		}
 		if (!ObjectUtils.isEmpty(files)) {
-			List<F> filteredFiles = filterFiles(files);
-			if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) {
-				rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));
-				List<F> 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<F> 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<F>
 					}
 				}
 				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<F>
 		}
 	}
 
+	private List<F> applyFilter(F[] files, boolean haveFilter, boolean filteringOneByOne, int maxFetchSize) {
+		List<F> 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<F> filteredFiles, F file) {
 		if (this.filter instanceof ReversibleFileListFilter) {
 			((ReversibleFileListFilter<F>) this.filter)
@@ -418,12 +448,8 @@ public abstract class AbstractInboundFileSynchronizer<F>
 				}
 				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<F>) this.filter).remove(remoteFile);
+			else {
+				resetFilterIfNecessary(remoteFile);
 			}
 		}
 		else if (this.logger.isWarnEnabled()) {
@@ -434,6 +460,16 @@ public abstract class AbstractInboundFileSynchronizer<F>
 		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<F>) this.filter).remove(remoteFile);
+		}
+	}
+
 	private boolean copyRemoteContentToLocalFile(Session<F> session, String remoteFilePath, File localFile) {
 		boolean renamed;
 		String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix;

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<String> compo =
+				new CompositeFileListFilter<>(Collections.singletonList(new FileListFilter<String>() {
+
+			@Override
+			public List<String> 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<String> compo =
+				new CompositeFileListFilter<>(Collections.singletonList(new FileListFilter<String>() {
+
+			@Override
+			public List<String> 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<String> compo =
+				new CompositeFileListFilter<>(Collections.singletonList(s -> null));
+		assertThat(compo.supportsSingleFileFiltering()).isFalse();
+		compo.close();
+	}
+
 }

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();
 	}
 
 }

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<String> 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<byte[]> received = (Message<byte[]>) 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<String> {
 
 		ConcurrentHashMap<String, String> metadataMap = new ConcurrentHashMap<>();
 
+		protected Streamer(RemoteFileTemplate<String> template) {
+			super(template, null);
+			doSetFilter(null);
+		}
+
 		protected Streamer(RemoteFileTemplate<String> template, Comparator<String> comparator) {
 			super(template, comparator);
 			doSetFilter(new StringPersistentFileListFilter(new SimpleMetadataStore(this.metadataMap), "streamer"));

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<String> sync = createLimitingSynchronizer(count);
 		AbstractInboundFileSynchronizingMessageSource<String> 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<String> 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<String> sync = createLimitingSynchronizer(count);
+		sync.setFilter(null);
+		AbstractInboundFileSynchronizingMessageSource<String> 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<String> sync = createLimitingSynchronizer(count);
+		ChainFileListFilter<String> cflf = new ChainFileListFilter<>();
+		cflf.addFilter(new AcceptOnceFileListFilter<>());
+		cflf.addFilter(fs -> Stream.of(fs).collect(Collectors.toList()));
+		sync.setFilter(cflf);
+		AbstractInboundFileSynchronizingMessageSource<String> 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<String> 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<String> 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<String> source = createSource(count);
 		source.setUseWatchService(true);
@@ -166,6 +224,7 @@ public class AbstractRemoteFileSynchronizerTests {
 
 	private AbstractInboundFileSynchronizingMessageSource<String> createSource(
 			AbstractInboundFileSynchronizer<String> sync) {
+
 		AbstractInboundFileSynchronizingMessageSource<String> source =
 				new AbstractInboundFileSynchronizingMessageSource<String>(sync) {
 
@@ -204,7 +263,7 @@ public class AbstractRemoteFileSynchronizerTests {
 
 			@Override
 			protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile,
-					File localDirectory, Session<String> session) throws IOException {
+					File localDirectory, Session<String> session) {
 				count.incrementAndGet();
 				return true;
 			}
@@ -227,40 +286,44 @@ public class AbstractRemoteFileSynchronizerTests {
 
 	private class StringSession implements Session<String> {
 
+		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;
 		}
 

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;
 	}

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.

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;
 	}

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 <<metadata-store>> 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.

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 <<remote-persistent-flf>>.
+
 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 <<metadata-store>>) 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 (<<ftp-streaming-java,later in the document>>) 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 <<remote-persistent-flf>>.
+
 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 <<ftp-max-fetch>> for more information.

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 <<remote-persistent-flf>>.
 
 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 <<sftp-streaming-java-config,shown later>> shows one technique to remove the remote file after processing, avoiding duplicates.
 
+For more information about the `SftpPersistentAcceptOnceFileListFilter`, and how it is used, see <<remote-persistent-flf>>.
+
 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 <<sftp-max-fetch>> for more information.

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 <<remote-persistent-flf>> 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.