GH-3488: Fix Persistent Filters with Recursion

Resolves #3488

Resolves two problems:

- When changes are made deep in the directory tree, they were not detected because
  the directory is in the metadata store and only passes the filter if a file
  immediately under it is changed, changing the directory's timestamp.

This is solved by subclassing `AbstractDirectoryAwareFileListFilter`, allowing its
`alwaysAcceptDirectories` property to be set.

- Only the filename was used as a metadata key; causing problems if a file with the
  same name appears multiple times in the tree.

This is solved with a new property on `AbstractDirectoryAwareFileListFilter` used by
the gateways to determine whether to filter the raw file names returned by the session
(previous behavior) or the full path relative to the root directory.

**cherry-pick to 5.4.x, 5.3.x**

* Some code style clean up

# Conflicts:
#	spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java
#	src/reference/asciidoc/whats-new.adoc

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -21,14 +21,19 @@
  * This permits, for example, pattern matching on just files when using recursion
  * to examine a directory tree.
  *
+ * @param <F> the file type.
+ *
  * @author Gary Russell
+ *
  * @since 5.0
  *
  */
 public abstract class AbstractDirectoryAwareFileListFilter<F> extends AbstractFileListFilter<F> {
 
 	private boolean alwaysAcceptDirectories;
 
+	private boolean forRecursion;
+
 	/**
 	 * Set to true so that filters that support this feature can unconditionally pass
 	 * directories; default false.
@@ -38,6 +43,22 @@ public void setAlwaysAcceptDirectories(boolean alwaysAcceptDirectories) {
 		this.alwaysAcceptDirectories = alwaysAcceptDirectories;
 	}
 
+	@Override
+	public boolean isForRecursion() {
+		return this.forRecursion;
+	}
+
+	/**
+	 * Set to true to inform a recursive gateway operation to use the full file path as
+	 * the metadata key. Also sets {@link #alwaysAcceptDirectories}.
+	 * @param forRecursion true to use the full path.
+	 * @since 5.3.6
+	 */
+	public void setForRecursion(boolean forRecursion) {
+		this.forRecursion = forRecursion;
+		this.alwaysAcceptDirectories = forRecursion;
+	}
+
 	protected boolean alwaysAccept(F file) {
 		return file != null && this.alwaysAcceptDirectories && isDirectory(file);
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2021 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,13 +31,15 @@
  * Files are deemed as already 'seen' if they exist in the store and have the
  * same modified time as the current file.
  *
+ * @param <F> the file type.
+ *
  * @author Gary Russell
  * @author Artem Bilan
  *
  * @since 3.0
  *
  */
-public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractFileListFilter<F>
+public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractDirectoryAwareFileListFilter<F>
 		implements ReversibleFileListFilter<F>, ResettableFileListFilter<F>, Closeable {
 
 	protected final ConcurrentMetadataStore store; // NOSONAR
@@ -73,6 +75,9 @@ public void setFlushOnUpdate(boolean flushOnUpdate) {
 
 	@Override
 	public boolean accept(F file) {
+		if (alwaysAccept(file)) {
+			return true;
+		}
 		String key = buildKey(file);
 		String newValue = value(file);
 		String oldValue = this.store.putIfAbsent(key, newValue);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -56,16 +56,23 @@ public class CompositeFileListFilter<F>
 
 	private boolean allSupportAccept = true;
 
+	private boolean oneIsForRecursion;
+
 
 	public CompositeFileListFilter() {
 		this.fileFilters = new LinkedHashSet<>();
 	}
 
 	public CompositeFileListFilter(Collection<? extends FileListFilter<F>> fileFilters) {
 		this.fileFilters = new LinkedHashSet<>(fileFilters);
-		this.allSupportAccept = fileFilters.stream().allMatch(FileListFilter<F>::supportsSingleFileFiltering);
+		this.allSupportAccept = fileFilters.stream().allMatch(FileListFilter::supportsSingleFileFiltering);
+		this.oneIsForRecursion = fileFilters.stream().anyMatch(FileListFilter::isForRecursion);
 	}
 
+	@Override
+	public boolean isForRecursion() {
+		return this.oneIsForRecursion;
+	}
 
 	@Override
 	public void close() throws IOException {
@@ -77,7 +84,6 @@ public void close() throws IOException {
 	}
 
 	public CompositeFileListFilter<F> addFilter(FileListFilter<F> filter) {
-		this.allSupportAccept &= filter.supportsSingleFileFiltering();
 		return addFilters(Collections.singletonList(filter));
 	}
 
@@ -114,11 +120,10 @@ public CompositeFileListFilter<F> addFilters(Collection<? extends FileListFilter
 					throw new IllegalStateException(e);
 				}
 			}
+			this.allSupportAccept &= elf.supportsSingleFileFiltering();
+			this.oneIsForRecursion |= elf.isForRecursion();
 		}
 		this.fileFilters.addAll(filtersToAdd);
-		if (this.allSupportAccept) {
-			this.allSupportAccept = filtersToAdd.stream().allMatch(FileListFilter<F>::supportsSingleFileFiltering);
-		}
 		return this;
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -64,4 +64,13 @@ default boolean supportsSingleFileFiltering() {
 		return false;
 	}
 
+	/**
+	 * Return true if this filter is being used for recursion.
+	 * @return whether or not to filter based on the full path.
+	 * @since 5.3.6
+	 */
+	default boolean isForRecursion() {
+		return false;
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2021 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.
@@ -23,6 +23,7 @@
 
 /**
  * @author Gary Russell
+ *
  * @since 3.0
  *
  */
@@ -52,5 +53,9 @@ protected boolean fileStillExists(File file) {
 		return file.exists();
 	}
 
+	@Override
+	protected boolean isDirectory(File file) {
+		return file.isDirectory();
+	}
 
 }

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

@@ -97,6 +97,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	 */
 	private FileListFilter<F> filter;
 
+	private boolean filterAfterEnhancement;
+
 	/**
 	 * A {@link FileListFilter} that runs against the <em>local</em> file system view when
 	 * using MPUT.
@@ -402,6 +404,15 @@ public void setCharset(String charset) {
 	 */
 	public void setFilter(FileListFilter<F> filter) {
 		this.filter = filter;
+		this.filterAfterEnhancement = filter != null
+				&& filter.isForRecursion()
+				&& filter.supportsSingleFileFiltering()
+				&& this.options.contains(Option.RECURSIVE);
+		if (filter != null && !filter.isForRecursion()) {
+			this.logger.warn("When using recursion, you will normally want to set the filter's "
+					+ "'forRecursion' property; otherwise files added deep into the "
+					+ "directory tree may not be detected");
+		}
 	}
 
 	/**
@@ -945,12 +956,19 @@ private List<F> listFilesInRemoteDir(Session<F> session, String directory, Strin
 		List<F> lsFiles = new ArrayList<>();
 		String remoteDirectory = buildRemotePath(directory, subDirectory);
 
-		F[] files = session.list(remoteDirectory);
-		boolean recursion = this.options.contains(Option.RECURSIVE);
+		F[] list = session.list(remoteDirectory);
+		List<F> files;
+		if (!this.filterAfterEnhancement) {
+			files = filterFiles(list);
+		}
+		else {
+			files = Arrays.asList(list);
+		}
 		if (!ObjectUtils.isEmpty(files)) {
-			for (F file : filterFiles(files)) {
+			for (F file : files) {
 				if (file != null) {
-					processFile(session, directory, subDirectory, lsFiles, recursion, file);
+					processFile(session, directory, subDirectory, lsFiles, this.options.contains(Option.RECURSIVE),
+							file);
 				}
 			}
 		}
@@ -972,29 +990,51 @@ protected final List<F> filterFiles(F[] files) {
 		return (this.filter != null) ? this.filter.filterFiles(files) : Arrays.asList(files);
 	}
 
+	protected final F filterFile(F file) {
+		if (this.filter.accept(file)) {
+			return file;
+		}
+		else {
+			return null;
+		}
+	}
+
 	private void processFile(Session<F> session, String directory, String subDirectory, List<F> lsFiles,
 			boolean recursion, F file) throws IOException {
 
-		String fileName = getFilename(file);
-		String fileSep = this.remoteFileTemplate.getRemoteFileSeparator();
-		boolean isDots = ".".equals(fileName)
-				|| "..".equals(fileName)
-				|| fileName.endsWith(fileSep + ".")
-				|| fileName.endsWith(fileSep + "..");
-		if (this.options.contains(Option.SUBDIRS) || !isDirectory(file)) {
-			if (recursion && StringUtils.hasText(subDirectory) && (!isDots || this.options.contains(Option.ALL))) {
-				lsFiles.add(enhanceNameWithSubDirectory(file, subDirectory));
+		F fileToAdd = file;
+		if (recursion && StringUtils.hasText(subDirectory)) {
+			fileToAdd = enhanceNameWithSubDirectory(file, subDirectory);
+		}
+		if (this.filterAfterEnhancement) {
+			if (!this.filter.accept(fileToAdd)) {
+				return;
 			}
+		}
+		String fileName = getFilename(fileToAdd);
+		final boolean isDirectory = isDirectory(file);
+		boolean isDots = hasDots(fileName);
+		if ((this.options.contains(Option.SUBDIRS) || !isDirectory)
+				&& (!isDots || this.options.contains(Option.ALL))) {
+
 			else if (this.options.contains(Option.ALL) || !isDots) {
 				lsFiles.add(file);
 			}
 		}
-		if (recursion && isDirectory(file) && !isDots) {
-			lsFiles.addAll(listFilesInRemoteDir(session, directory,
-					subDirectory + fileName + fileSep));
+
+		if (recursion && isDirectory && !isDots) {
+			lsFiles.addAll(listFilesInRemoteDir(session, directory, fileName +
+					this.remoteFileTemplate.getRemoteFileSeparator()));
 		}
 	}
 
+	private boolean hasDots(String fileName) {
+		String fileSeparator = this.remoteFileTemplate.getRemoteFileSeparator();
+		return ".".equals(fileName)
+				|| "..".equals(fileName)
+				|| fileName.endsWith(fileSeparator + ".")
+				|| fileName.endsWith(fileSeparator + "..");
+	}
 	protected final List<File> filterMputFiles(File[] files) {
 		if (files == null) {
 			return Collections.emptyList();