Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4351: Add DiscardAwareFileListFilter

Browse Source 
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`
This commit is contained in:
Artem Bilan
2018-03-29 17:37:24 -04:00
committed by Gary Russell
parent 10805e5b8a
commit a48d54bed9
6 changed files with 135 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Path, WatchKey> pathKeys = new ConcurrentHashMap<>();
private final Set<File> filesToPoll = ConcurrentHashMap.newKeySet();
private WatchService watcher;
private Collection<File> initialFiles;
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 class FileReadingMessageSource extends IntegrationObjectSupport implement
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 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<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()) {

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

@@ -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 <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,7 +73,7 @@ public class CompositeFileListFilter<F> implements ReversibleFileListFilter<F>,
}
public CompositeFileListFilter<F> addFilter(FileListFilter<F> filter) {
return this.addFilters(Collections.singletonList(filter));
return addFilters(Collections.singletonList(filter));
}
/**
@@ -73,7 +81,7 @@ public class CompositeFileListFilter<F> implements ReversibleFileListFilter<F>,
* @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 class CompositeFileListFilter<F> implements ReversibleFileListFilter<F>,
* @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 class CompositeFileListFilter<F> implements ReversibleFileListFilter<F>,
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) {

33
spring-integration-file/src/main/java/org/springframework/integration/file/filters/DiscardAwareFileListFilter.java Normal file
View File

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

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

@@ -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.
* <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 class LastModifiedFileListFilter implements FileListFilter<File> {
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<File> {
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;
}

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

@@ -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<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));

11
src/reference/asciidoc/file.adoc
View File

@@ -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`: