INT-4351: Add DiscardAwareFileListFilter

JIRA: https://jira.spring.io/browse/INT-4351

The `WatchService` reacts to the events in the file system and keep
track ove the events until we poll them, e.g. via `listEligibleFiles()`
in the `WatchServiceDirectoryScanner`.
On the other hand the `SourcePollingChannelAdapter` calls the mentioned
`listEligibleFiles()` according its polling period.
At this moment the `FileListFilter` is applied to the polled files.
It may happen that `LastModifiedFileListFilter` can't accept too young
files yet and they are lost for the future consideration.

* To allow, for example, to retain young files by the
`LastModifiedFileListFilter` judgment for the future cycles add
`DiscardAwareFileListFilter` with the `DiscardCallback` support.
The `WatchServiceDirectoryScanner` now registers such a callback into
the filter and stores discarded files into the `filesToPoll` queue
for the future poll cycle.

**Cherry-pick to 5.0**

Fix compilation warnings

* Replace `DiscardCallback` with the plain `Consumer`
* Ensure uniqueness in the `WatchServiceDirectoryScanner` internal
queue via a `Set` implementation, since discard callback may be called
several times for the same file from the `CompositeFileListFilter`
according to its nature
* Add JavaDocs and Docs

Change `@since` to `5.0.5`

Author: artembilan

spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java

@@ -29,8 +29,8 @@
 import java.nio.file.WatchService;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.Comparator;
+import java.util.Iterator;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Queue;
@@ -49,6 +49,7 @@
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
+import org.springframework.integration.file.filters.DiscardAwareFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.filters.ResettableFileListFilter;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
@@ -435,12 +436,20 @@ private class WatchServiceDirectoryScanner extends DefaultDirectoryScanner imple
 
 		private final ConcurrentMap<Path, WatchKey> pathKeys = new ConcurrentHashMap<>();
 
-		private WatchService watcher;
+		private final Set<File> filesToPoll = ConcurrentHashMap.newKeySet();
 
-		private Collection<File> initialFiles;
+		private WatchService watcher;
 
 		private WatchEvent.Kind<?>[] kinds;
 
+		@Override
+		public void setFilter(FileListFilter<File> filter) {
+			if (filter instanceof DiscardAwareFileListFilter) {
+				((DiscardAwareFileListFilter<File>) filter).addDiscardCallback(this.filesToPoll::add);
+			}
+			super.setFilter(filter);
+		}
+
 		@Override
 		public void start() {
 			try {
@@ -456,9 +465,9 @@ public void start() {
 				this.kinds[i] = FileReadingMessageSource.this.watchEvents[i].kind;
 			}
 
-			final Set<File> initialFiles = walkDirectory(FileReadingMessageSource.this.directory.toPath(), null);
+			Set<File> initialFiles = walkDirectory(FileReadingMessageSource.this.directory.toPath(), null);
 			initialFiles.addAll(filesFromEvents());
-			this.initialFiles = initialFiles;
+			this.filesToPoll.addAll(initialFiles);
 		}
 
 		@Override
@@ -481,18 +490,22 @@ public boolean isRunning() {
 		@Override
 		protected File[] listEligibleFiles(File directory) {
 			Assert.state(this.watcher != null, "The WatchService has'nt been started");
-			if (this.initialFiles != null) {
-				File[] initial = this.initialFiles.toArray(new File[this.initialFiles.size()]);
-				this.initialFiles = null;
-				return initial;
+
+			Set<File> files = new LinkedHashSet<>();
+
+			for (Iterator<File> iterator = this.filesToPoll.iterator(); iterator.hasNext(); ) {
+				files.add(iterator.next());
+				iterator.remove();
 			}
-			Collection<File> files = filesFromEvents();
+
+			files.addAll(filesFromEvents());
+
 			return files.toArray(new File[files.size()]);
 		}
 
 		private Set<File> filesFromEvents() {
 			WatchKey key = this.watcher.poll();
-			Set<File> files = new LinkedHashSet<File>();
+			Set<File> files = new LinkedHashSet<>();
 			while (key != null) {
 				File parentDir = ((Path) key.watchable()).toAbsolutePath().toFile();
 				for (WatchEvent<?> event : key.pollEvents()) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -25,33 +25,41 @@
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.function.Consumer;
 
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
 
 /**
  * Simple {@link FileListFilter} that predicates its matches against <b>all</b> of the
  * configured {@link FileListFilter}.
+ * <p>
+ * Note: when {@link #discardCallback} is provided, it is populated to all the
+ * {@link DiscardAwareFileListFilter} delegates. In this case, since this filter
+ * matches the files against all delegates, the {@link #discardCallback} may be
+ * called several times for the same file.
+ *
  * @param <F> The type that will be filtered.
  *
  * @author Iwein Fuld
  * @author Josh Long
  * @author Gary Russell
  * @author Artem Bilan
- *
- *
  */
-public class CompositeFileListFilter<F> implements ReversibleFileListFilter<F>, ResettableFileListFilter<F>, Closeable {
+public class CompositeFileListFilter<F>
+		implements ReversibleFileListFilter<F>, ResettableFileListFilter<F>, DiscardAwareFileListFilter<F>, Closeable {
 
 	protected final Set<FileListFilter<F>> fileFilters; // NOSONAR
 
+	private Consumer<F> discardCallback;
+
 
 	public CompositeFileListFilter() {
-		this.fileFilters = new LinkedHashSet<FileListFilter<F>>();
+		this.fileFilters = new LinkedHashSet<>();
 	}
 
 	public CompositeFileListFilter(Collection<? extends FileListFilter<F>> fileFilters) {
-		this.fileFilters = new LinkedHashSet<FileListFilter<F>>(fileFilters);
+		this.fileFilters = new LinkedHashSet<>(fileFilters);
 	}
 
 
@@ -65,15 +73,15 @@ public void close() throws IOException {
 	}
 
 	public CompositeFileListFilter<F> addFilter(FileListFilter<F> filter) {
-		return this.addFilters(Collections.singletonList(filter));
+		return addFilters(Collections.singletonList(filter));
 	}
 
 	/**
 	 * @param filters one or more new filters to add
 	 * @return this CompositeFileFilter instance with the added filters
 	 * @see #addFilters(Collection)
 	 */
-	@SuppressWarnings("unchecked") //For JDK7
+	@SuppressWarnings("unchecked")
 	public CompositeFileListFilter<F> addFilters(FileListFilter<F>... filters) {
 		return addFilters(Arrays.asList(filters));
 	}
@@ -87,7 +95,10 @@ public CompositeFileListFilter<F> addFilters(FileListFilter<F>... filters) {
 	 * @return this CompositeFileListFilter instance with the added filters
 	 */
 	public CompositeFileListFilter<F> addFilters(Collection<? extends FileListFilter<F>> filtersToAdd) {
-		for (FileListFilter<? extends F> elf : filtersToAdd) {
+		for (FileListFilter<F> elf : filtersToAdd) {
+			if (elf instanceof DiscardAwareFileListFilter) {
+				((DiscardAwareFileListFilter<F>) elf).addDiscardCallback(this.discardCallback);
+			}
 			if (elf instanceof InitializingBean) {
 				try {
 					((InitializingBean) elf).afterPropertiesSet();
@@ -101,6 +112,17 @@ public CompositeFileListFilter<F> addFilters(Collection<? extends FileListFilter
 		return this;
 	}
 
+	@Override
+	public void addDiscardCallback(Consumer<F> discardCallback) {
+		this.discardCallback = discardCallback;
+		if (this.discardCallback != null) {
+			this.fileFilters
+					.stream()
+					.filter(DiscardAwareFileListFilter.class::isInstance)
+					.map(f -> (DiscardAwareFileListFilter<F>) f)
+					.forEach(f -> f.addDiscardCallback(discardCallback));
+		}
+	}
 
 	@Override
 	public List<F> filterFiles(F[] files) {

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.file.filters;
+
+import java.util.function.Consumer;
+
+/**
+ * The {@link FileListFilter} modification which can accept a {@link Consumer}
+ * which can be called when filter discards the file.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0.5
+ */
+public interface DiscardAwareFileListFilter<F> extends FileListFilter<F> {
+
+	void addDiscardCallback(Consumer<F> discardCallback);
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -20,28 +20,31 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 /**
  * The {@link FileListFilter} implementation to filter those files which
  * {@link File#lastModified()} is less than the {@link #age} in comparison
  * with the current time.
  * <p>
  * The resolution is done in seconds.
+ * <p>
+ * When {@link #discardCallback} is provided, it called for all the
+ * rejected files.
  *
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 4.2
  *
  */
-public class LastModifiedFileListFilter implements FileListFilter<File> {
+public class LastModifiedFileListFilter implements DiscardAwareFileListFilter<File> {
 
 	private static final long DEFAULT_AGE = 60;
 
 	private volatile long age = DEFAULT_AGE;
 
-	public long getAge() {
-		return this.age;
-	}
+	private Consumer<File> discardCallback;
 
 	public LastModifiedFileListFilter() {
 	}
@@ -67,6 +70,10 @@ public void setAge(long age) {
 		setAge(age, TimeUnit.SECONDS);
 	}
 
+	public long getAge() {
+		return this.age;
+	}
+
 	/**
 	 * Set the age that files have to be before being passed by this filter.
 	 * If {@link File#lastModified()} plus age is greater than the current time, the file
@@ -79,14 +86,22 @@ public void setAge(long age, TimeUnit unit) {
 		this.age = unit.toSeconds(age);
 	}
 
+	@Override
+	public void addDiscardCallback(Consumer<File> discardCallback) {
+		this.discardCallback = discardCallback;
+	}
+
 	@Override
 	public List<File> filterFiles(File[] files) {
-		List<File> list = new ArrayList<File>();
+		List<File> list = new ArrayList<>();
 		long now = System.currentTimeMillis() / 1000;
 		for (File file : files) {
 			if (file.lastModified() / 1000 + this.age <= now) {
 				list.add(file);
 			}
+			else if (this.discardCallback != null) {
+				this.discardCallback.accept(file);
+			}
 		}
 		return list;
 	}

spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java

@@ -39,7 +39,9 @@
 import org.junit.rules.TemporaryFolder;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.file.filters.ChainFileListFilter;
 import org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter;
+import org.springframework.integration.file.filters.LastModifiedFileListFilter;
 import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
@@ -75,6 +77,7 @@ public void setUp() throws IOException {
 	}
 
 	@Test
+	@SuppressWarnings("unchecked")
 	public void testWatchServiceDirectoryScanner() throws Exception {
 		FileReadingMessageSource fileReadingMessageSource = new FileReadingMessageSource();
 		fileReadingMessageSource.setDirectory(folder.getRoot());
@@ -86,7 +89,7 @@ public void testWatchServiceDirectoryScanner() throws Exception {
 
 		final CountDownLatch removeFileLatch = new CountDownLatch(1);
 
-		FileSystemPersistentAcceptOnceFileListFilter filter =
+		FileSystemPersistentAcceptOnceFileListFilter fileSystemPersistentAcceptOnceFileListFilter =
 				new FileSystemPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "test") {
 
 					@Override
@@ -97,14 +100,24 @@ public boolean remove(File fileToRemove) {
 
 				};
 
-		fileReadingMessageSource.setFilter(filter);
+		LastModifiedFileListFilter fileLastModifiedFileListFilter = new LastModifiedFileListFilter();
+
+		ChainFileListFilter<File> fileChainFileListFilter = new ChainFileListFilter<>();
+		fileChainFileListFilter.addFilters(fileLastModifiedFileListFilter, fileSystemPersistentAcceptOnceFileListFilter);
+
+		fileReadingMessageSource.setFilter(fileChainFileListFilter);
 		fileReadingMessageSource.afterPropertiesSet();
 		fileReadingMessageSource.start();
 		DirectoryScanner scanner = fileReadingMessageSource.getScanner();
 		assertThat(scanner.getClass().getName(),
 				containsString("FileReadingMessageSource$WatchServiceDirectoryScanner"));
 
+		// Files are skipped by the LastModifiedFileListFilter
 		List<File> files = scanner.listFiles(folder.getRoot());
+		assertEquals(0, files.size());
+		// Consider all the files as one day old
+		fileLastModifiedFileListFilter.setAge(-60 * 60 * 24);
+		files = scanner.listFiles(folder.getRoot());
 		assertEquals(3, files.size());
 		assertTrue(files.contains(top1));
 		assertTrue(files.contains(foo1));

src/reference/asciidoc/file.adoc

@@ -130,6 +130,17 @@ For this purpose all the XML components for file handling (local and remote), al
         auto-startup="false"/>
 ----
 
+Starting with _version 5.0.5_, a `DiscardAwareFileListFilter` is provided for implementations when there is an interest in the event of the rejected files.
+For this purpose such a filter implementation should be supplied with a callback via `addDiscardCallback(Consumer<File>)`.
+In the Framework this functionality is used from the `FileReadingMessageSource.WatchServiceDirectoryScanner` in combination with `LastModifiedFileListFilter`.
+Unlike the regular `DirectoryScanner`, the `WatchService` provides files for processing according the events on the target file system.
+At the moment of polling an internal queue with those files, the `LastModifiedFileListFilter` may discard them because they are too young in regards to its configured `age`.
+Therefore we lose the file for the future possible considerations.
+The discard callback hook allows us to retain the file in the internal queue, so it is available to be checked against the `age` in the subsequent polls.
+The `CompositeFileListFilter` also implements a `DiscardAwareFileListFilter` and populates provided discard callback to all its `DiscardAwareFileListFilter` delegates.
+
+NOTE: Since `CompositeFileListFilter` matches the files against all delegates, the `discardCallback` may be called several times for the same file.
+
 *Message Headers*
 
 Starting with _version 5.0_ the `FileReadingMessageSource`, in addition to the `payload` as a polled `File`, populates these headers to the outbound `Message`: