GH-2777: Remote File Filter Improvements

Resolves #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

Author: garyrussell

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 @@
  *
  * @author Mark Fisher
  * @author Iwein Fuld
+ * @author Gary Russell
  */
 public abstract class AbstractFileListFilter<F> implements FileListFilter<F> {
 
@@ -42,11 +43,17 @@ public final List<F> filterFiles(F[] files) {
 		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 @@
  * 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 @@ protected void flushIfNeeded() {
 			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 List<F> filterFiles(F[] files) {
 		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.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,13 +54,16 @@ public class CompositeFileListFilter<F>
 
 	private Consumer<F> discardCallback;
 
+	private boolean allSupportAccept = true;
+
 
 	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);
 	}
 
 
@@ -73,6 +77,7 @@ public void close() throws IOException {
 	}
 
 	public CompositeFileListFilter<F> addFilter(FileListFilter<F> filter) {
+		this.allSupportAccept &= filter.supportsSingleFileFiltering();
 		return addFilters(Collections.singletonList(filter));
 	}
 
@@ -81,9 +86,11 @@ public CompositeFileListFilter<F> addFilter(FileListFilter<F> filter) {
 	 * @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 CompositeFileListFilter<F> addFilters(Collection<? extends FileListFilter
 			}
 		}
 		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 List<F> filterFiles(F[] files) {
 		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 @@
 
 /**
  * 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 @@
  *
  * @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,16 +105,16 @@ public long getAge() {
 	}
 
 	@Override
-	public void addDiscardCallback(@Nullable Consumer<File> discardCallback) {
-		this.discardCallback = discardCallback;
+	public void addDiscardCallback(@Nullable Consumer<File> discardCallbackToSet) {
+		this.discardCallback = discardCallbackToSet;
 	}
 
 	@Override
 	public List<File> filterFiles(File[] files) {
 		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 @@ else if (this.discardCallback != null) {
 		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;
+	}
+
+
 }