diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 029e444e0c..01bb835bf9 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -19,6 +19,7 @@ package org.springframework.integration.file; import java.io.File; import java.io.FileFilter; import java.io.IOException; +import java.util.Comparator; import java.util.HashSet; import java.util.List; import java.util.Queue; @@ -28,6 +29,7 @@ import java.util.concurrent.PriorityBlockingQueue; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.core.io.Resource; +import org.springframework.integration.aggregator.Resequencer; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessagingException; import org.springframework.integration.message.GenericMessage; @@ -35,9 +37,9 @@ import org.springframework.integration.message.MessageSource; import org.springframework.util.Assert; /** - * PollableSource that creates messages from a file system directory. To prevent - * messages for certain files, you may supply a {@link FileListFilter}. By - * default, an {@link AcceptOnceFileListFilter} is used. It ensures files are + * {@link MessageSource} that creates messages from a file system directory. To + * prevent messages for certain files, you may supply a {@link FileListFilter}. + * By default, an {@link AcceptOnceFileListFilter} is used. It ensures files are * picked up only once from the directory. *

* A common problem with reading files is that a file may be detected before it @@ -48,6 +50,11 @@ import org.springframework.util.Assert; * with the default {@link AcceptOnceFileListFilter} would allow for this. See * {@link CompositeFileFilter} for a way to do this. *

+ * A {@link Comparator} can be used to ensure internal ordering of the Files in + * a {@link PriorityBlockingQueue}. This does not provide the same guarantees as + * a {@link Resequencer}, but in cases where writing files and failure + * downstream are rare it might be sufficient. + *

* FileReadingMessageSource is fully thread-safe under concurrent receives and * message delivery callbacks. * @@ -55,6 +62,8 @@ import org.springframework.util.Assert; */ public class FileReadingMessageSource implements MessageSource { + private static final int INTERNAL_QUEUE_CAPACITY = 5; + private static final Log logger = LogFactory.getLog(FileReadingMessageSource.class); private volatile File inputDirectory; @@ -64,10 +73,28 @@ public class FileReadingMessageSource implements MessageSource { * {@link ConcurrentModificationException} in Java 5. There is no locking * around the queue, so there is also no iteration. */ - private final Queue toBeReceived = new PriorityBlockingQueue(); + private final Queue toBeReceived; private volatile FileListFilter filter = new AcceptOnceFileListFilter(); + /** + * Creates a FileReadingMessageSource with a naturally ordered queue. + */ + public FileReadingMessageSource() { + toBeReceived = new PriorityBlockingQueue(INTERNAL_QUEUE_CAPACITY); + } + + /** + * Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue} + * ordered with the passed in {@link Comparator} + * + * No guarantees about file delivery order can be made under concurrent + * access. + */ + public FileReadingMessageSource(Comparator receptionOrderComparator) { + toBeReceived = new PriorityBlockingQueue(INTERNAL_QUEUE_CAPACITY, receptionOrderComparator); + } + public void setInputDirectory(Resource inputDirectory) { Assert.notNull(inputDirectory, "inputDirectory cannot be null"); Assert.isTrue(inputDirectory.exists(), inputDirectory + " doesn't exist."); @@ -87,7 +114,7 @@ public class FileReadingMessageSource implements MessageSource { * and duplication concerns. If multiple filters are required a * {@link CompositeFileListFilter} can be used to group them together *

- * Note that the supplied filter must be thread safe. + * Note that the supplied filter must be thread safe.. */ public void setFilter(FileListFilter filter) { Assert.notNull(filter, "'filter' should not be null"); diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java index 31d22aac76..006ceebfd4 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java @@ -19,8 +19,10 @@ import static org.easymock.classextension.EasyMock.*; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import java.io.File; +import java.util.Comparator; import org.junit.Before; import org.junit.Test; @@ -41,7 +43,9 @@ public class FileReadingMessageSourceTests { private File fileMock = createMock(File.class); - private Object[] allMocks = new Object[] { inputDirectoryMock, fileMock, inputDirectoryResourceMock }; + private Comparator comparator = createMock(Comparator.class); + + private Object[] allMocks = new Object[] { inputDirectoryMock, fileMock, inputDirectoryResourceMock, comparator }; public void prepResource() throws Exception { expect(inputDirectoryResourceMock.exists()).andReturn(true).anyTimes(); @@ -49,11 +53,11 @@ public class FileReadingMessageSourceTests { expect(inputDirectoryMock.canRead()).andReturn(true); replay(inputDirectoryResourceMock, inputDirectoryMock); } - + @Before public void initialize() throws Exception { prepResource(); - this.source = new FileReadingMessageSource(); + this.source = new FileReadingMessageSource(comparator); source.setInputDirectory(inputDirectoryResourceMock); reset(allMocks); } @@ -68,8 +72,7 @@ public class FileReadingMessageSourceTests { @Test public void requeueOnFailure() throws Exception { - expect(inputDirectoryMock.listFiles()).andReturn(new File[] { fileMock }); - expect(inputDirectoryMock.listFiles()).andReturn(new File[] {}); + expect(inputDirectoryMock.listFiles()).andReturn(new File[] { fileMock }).times(2); replay(allMocks); Message received = source.receive(); assertNotNull(received); @@ -80,8 +83,7 @@ public class FileReadingMessageSourceTests { @Test public void noDuplication() throws Exception { - expect(inputDirectoryMock.listFiles()).andReturn(new File[] { fileMock }); - expect(inputDirectoryMock.listFiles()).andReturn(new File[] {}); + expect(inputDirectoryMock.listFiles()).andReturn(new File[] { fileMock }).times(2); replay(allMocks); Message received = source.receive(); assertNotNull(received); @@ -89,4 +91,27 @@ public class FileReadingMessageSourceTests { assertNull(source.receive()); verify(allMocks); } + + @Test + public void orderedReception() throws Exception { + File file1 = createMock(File.class); + File file2 = createMock(File.class); + File file3 = createMock(File.class); + + //record the comparator to reverse order the files + expect(comparator.compare(file1, file2)).andReturn(1).anyTimes(); + expect(comparator.compare(file1, file3)).andReturn(1).anyTimes(); + expect(comparator.compare(file2, file3)).andReturn(1).anyTimes(); + expect(comparator.compare(file2, file1)).andReturn(-1).anyTimes(); + expect(comparator.compare(file3, file1)).andReturn(-1).anyTimes(); + expect(comparator.compare(file3, file2)).andReturn(-1).anyTimes(); + + expect(inputDirectoryMock.listFiles()).andReturn(new File[] { file2, file3, file1 }).anyTimes(); + replay(allMocks); + assertSame(file3, source.receive().getPayload()); + assertSame(file2, source.receive().getPayload()); + assertSame(file1, source.receive().getPayload()); + assertNull(source.receive()); + verify(allMocks); + } }