diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index a4f8f66c65..569ff5e0ee 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -29,8 +29,8 @@ import java.nio.file.WatchKey; 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.aggregator.ResequencingMessageGroupProces 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 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement private final ConcurrentMap pathKeys = new ConcurrentHashMap<>(); + private final Set filesToPoll = ConcurrentHashMap.newKeySet(); + private WatchService watcher; - private Collection initialFiles; - private WatchEvent.Kind[] kinds; + @Override + public void setFilter(FileListFilter filter) { + if (filter instanceof DiscardAwareFileListFilter) { + ((DiscardAwareFileListFilter) filter).addDiscardCallback(this.filesToPoll::add); + } + super.setFilter(filter); + } + @Override public void start() { try { @@ -456,9 +465,9 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement this.kinds[i] = FileReadingMessageSource.this.watchEvents[i].kind; } - final Set initialFiles = walkDirectory(FileReadingMessageSource.this.directory.toPath(), null); + Set initialFiles = walkDirectory(FileReadingMessageSource.this.directory.toPath(), null); initialFiles.addAll(filesFromEvents()); - this.initialFiles = initialFiles; + this.filesToPoll.addAll(initialFiles); } @Override @@ -481,18 +490,22 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement @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 files = new LinkedHashSet<>(); + + for (Iterator iterator = this.filesToPoll.iterator(); iterator.hasNext(); ) { + files.add(iterator.next()); + iterator.remove(); } - Collection files = filesFromEvents(); + + files.addAll(filesFromEvents()); + return files.toArray(new File[files.size()]); } private Set filesFromEvents() { WatchKey key = this.watcher.poll(); - Set files = new LinkedHashSet(); + Set files = new LinkedHashSet<>(); while (key != null) { File parentDir = ((Path) key.watchable()).toAbsolutePath().toFile(); for (WatchEvent event : key.pollEvents()) { 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 2af226e57a..339759fb45 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 @@ -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,6 +25,7 @@ import java.util.Collections; 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; @@ -32,26 +33,33 @@ import org.springframework.util.Assert; /** * Simple {@link FileListFilter} that predicates its matches against all of the * configured {@link FileListFilter}. + *

+ * 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 The type that will be filtered. * * @author Iwein Fuld * @author Josh Long * @author Gary Russell * @author Artem Bilan - * - * */ -public class CompositeFileListFilter implements ReversibleFileListFilter, ResettableFileListFilter, Closeable { +public class CompositeFileListFilter + implements ReversibleFileListFilter, ResettableFileListFilter, DiscardAwareFileListFilter, Closeable { protected final Set> fileFilters; // NOSONAR + private Consumer discardCallback; + public CompositeFileListFilter() { - this.fileFilters = new LinkedHashSet>(); + this.fileFilters = new LinkedHashSet<>(); } public CompositeFileListFilter(Collection> fileFilters) { - this.fileFilters = new LinkedHashSet>(fileFilters); + this.fileFilters = new LinkedHashSet<>(fileFilters); } @@ -65,7 +73,7 @@ public class CompositeFileListFilter implements ReversibleFileListFilter, } public CompositeFileListFilter addFilter(FileListFilter filter) { - return this.addFilters(Collections.singletonList(filter)); + return addFilters(Collections.singletonList(filter)); } /** @@ -73,7 +81,7 @@ public class CompositeFileListFilter implements ReversibleFileListFilter, * @return this CompositeFileFilter instance with the added filters * @see #addFilters(Collection) */ - @SuppressWarnings("unchecked") //For JDK7 + @SuppressWarnings("unchecked") public CompositeFileListFilter addFilters(FileListFilter... filters) { return addFilters(Arrays.asList(filters)); } @@ -87,7 +95,10 @@ public class CompositeFileListFilter implements ReversibleFileListFilter, * @return this CompositeFileListFilter instance with the added filters */ public CompositeFileListFilter addFilters(Collection> filtersToAdd) { - for (FileListFilter elf : filtersToAdd) { + for (FileListFilter elf : filtersToAdd) { + if (elf instanceof DiscardAwareFileListFilter) { + ((DiscardAwareFileListFilter) elf).addDiscardCallback(this.discardCallback); + } if (elf instanceof InitializingBean) { try { ((InitializingBean) elf).afterPropertiesSet(); @@ -101,6 +112,17 @@ public class CompositeFileListFilter implements ReversibleFileListFilter, return this; } + @Override + public void addDiscardCallback(Consumer discardCallback) { + this.discardCallback = discardCallback; + if (this.discardCallback != null) { + this.fileFilters + .stream() + .filter(DiscardAwareFileListFilter.class::isInstance) + .map(f -> (DiscardAwareFileListFilter) f) + .forEach(f -> f.addDiscardCallback(discardCallback)); + } + } @Override public List filterFiles(F[] files) { 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 new file mode 100644 index 0000000000..86334faae7 --- /dev/null +++ b/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 extends FileListFilter { + + void addDiscardCallback(Consumer discardCallback); + +} 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 8dfa08c8b8..a0d5116ef1 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 @@ -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,6 +20,7 @@ import java.io.File; 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 @@ -27,21 +28,23 @@ import java.util.concurrent.TimeUnit; * with the current time. *

* The resolution is done in seconds. + *

+ * 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 { +public class LastModifiedFileListFilter implements DiscardAwareFileListFilter { private static final long DEFAULT_AGE = 60; private volatile long age = DEFAULT_AGE; - public long getAge() { - return this.age; - } + private Consumer discardCallback; public LastModifiedFileListFilter() { } @@ -67,6 +70,10 @@ public class LastModifiedFileListFilter implements FileListFilter { 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 class LastModifiedFileListFilter implements FileListFilter { this.age = unit.toSeconds(age); } + @Override + public void addDiscardCallback(Consumer discardCallback) { + this.discardCallback = discardCallback; + } + @Override public List filterFiles(File[] files) { - List list = new ArrayList(); + List 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; } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java index 2d20c92940..5a1bd3c55f 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java @@ -39,7 +39,9 @@ import org.junit.Test; 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 class WatchServiceDirectoryScannerTests { } @Test + @SuppressWarnings("unchecked") public void testWatchServiceDirectoryScanner() throws Exception { FileReadingMessageSource fileReadingMessageSource = new FileReadingMessageSource(); fileReadingMessageSource.setDirectory(folder.getRoot()); @@ -86,7 +89,7 @@ public class WatchServiceDirectoryScannerTests { final CountDownLatch removeFileLatch = new CountDownLatch(1); - FileSystemPersistentAcceptOnceFileListFilter filter = + FileSystemPersistentAcceptOnceFileListFilter fileSystemPersistentAcceptOnceFileListFilter = new FileSystemPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "test") { @Override @@ -97,14 +100,24 @@ public class WatchServiceDirectoryScannerTests { }; - fileReadingMessageSource.setFilter(filter); + LastModifiedFileListFilter fileLastModifiedFileListFilter = new LastModifiedFileListFilter(); + + ChainFileListFilter 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 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)); diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 2028fc70e0..a7bffd541e 100644 --- a/src/reference/asciidoc/file.adoc +++ b/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)`. +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`: