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

OPEN - issue INT-583: FileReadingMessageSource rescanning directory every pass

Browse Source 
http://jira.springframework.org/browse/INT-583

Fixed issue, added flag to go back to old behavior. Refactored test to Mockito.
This commit is contained in:
Iwein Fuld
2009-02-23 16:19:36 +00:00
parent 0d960df263
commit a0da740a3d
4 changed files with 71 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -54,8 +54,8 @@ import org.springframework.util.Assert;
* a {@link Resequencer}, but in cases where writing files and failure
* downstream are rare it might be sufficient.
* <p/>
* FileReadingMessageSource is fully thread-safe under concurrent receives and
* message delivery callbacks.
* FileReadingMessageSource is fully thread-safe under concurrent
* <code>receive()</code> invocations and message delivery callbacks.
*
* @author Iwein Fuld
*/
@@ -76,6 +76,8 @@ public class FileReadingMessageSource implements MessageSource<File> {
private volatile FileListFilter filter = new AcceptOnceFileListFilter();
private boolean scanEachPoll = false;
/**
* Creates a FileReadingMessageSource with a naturally ordered queue.
*/
@@ -109,8 +111,8 @@ public class FileReadingMessageSource implements MessageSource<File> {
/**
* Sets a {@link FileListFilter}. By default a
* {@link AcceptOnceFileListFilter} with no bounds is used. In most cases a
* customized {@link FileListFilter} will be needed to deal with modification
* and duplication concerns. If multiple filters are required a
* customized {@link FileListFilter} will be needed to deal with
* modification and duplication concerns. If multiple filters are required a
* {@link CompositeFileListFilter} can be used to group them together
* <p/>
* <b>The supplied filter must be thread safe.</b>.
@@ -120,9 +122,23 @@ public class FileReadingMessageSource implements MessageSource<File> {
this.filter = filter;
}
/**
* Optional. Set this flag if you want to make sure the internal queue is
* refreshed with the latest content of the input directory on each poll.
* <p/>
* By default this implementation will empty its queue before looking at the
* directory again. In cases where order
*/
public void setScanEachPoll(boolean scanEachPoll) {
this.scanEachPoll = scanEachPoll;
}
public Message<File> receive() throws MessagingException {
refreshQueue();
Message<File> message = null;
// rescan only if needed or explicitly configured
if (toBeReceived.isEmpty() || scanEachPoll) {
scanInputDirectory();
}
File file = toBeReceived.poll();
// we can't rely on isEmpty for concurrency reasons
if (file != null) {
@@ -134,7 +150,7 @@ public class FileReadingMessageSource implements MessageSource<File> {
return message;
}
private void refreshQueue() {
private void scanInputDirectory() {
List<File> filteredFiles = filter.filterFiles((inputDirectory.listFiles()));
Set<File> freshFiles = new HashSet<File>(filteredFiles);
if (!freshFiles.isEmpty()) {