diff --git a/org.springframework.integration.file/.classpath b/org.springframework.integration.file/.classpath
index 021d1c4629..81bc0ff0f5 100644
--- a/org.springframework.integration.file/.classpath
+++ b/org.springframework.integration.file/.classpath
@@ -13,6 +13,8 @@
+
+
diff --git a/org.springframework.integration.file/ivy.xml b/org.springframework.integration.file/ivy.xml
index b725ba28d4..d6d30ed2d6 100644
--- a/org.springframework.integration.file/ivy.xml
+++ b/org.springframework.integration.file/ivy.xml
@@ -27,6 +27,7 @@
+
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 0fa09c8f5d..d5f9a9f386 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
@@ -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.
*
- * FileReadingMessageSource is fully thread-safe under concurrent receives and
- * message delivery callbacks.
+ * FileReadingMessageSource is fully thread-safe under concurrent
+ * receive() invocations and message delivery callbacks.
*
* @author Iwein Fuld
*/
@@ -76,6 +76,8 @@ public class FileReadingMessageSource implements MessageSource {
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 {
/**
* 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
*
* The supplied filter must be thread safe..
@@ -120,9 +122,23 @@ public class FileReadingMessageSource implements MessageSource {
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.
+ *
+ * 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 receive() throws MessagingException {
- refreshQueue();
Message 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 {
return message;
}
- private void refreshQueue() {
+ private void scanInputDirectory() {
List filteredFiles = filter.filterFiles((inputDirectory.listFiles()));
Set freshFiles = new HashSet(filteredFiles);
if (!freshFiles.isEmpty()) {
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 006ceebfd4..38979b7f85 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
@@ -15,7 +15,7 @@
*/
package org.springframework.integration.file;
-import static org.easymock.classextension.EasyMock.*;
+import static org.mockito.Mockito.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
@@ -26,6 +26,9 @@ import java.util.Comparator;
import org.junit.Before;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnit44Runner;
import org.springframework.core.io.Resource;
import org.springframework.integration.core.Message;
@@ -33,25 +36,27 @@ import org.springframework.integration.core.Message;
* @author Iwein Fuld
*/
@SuppressWarnings("unchecked")
+@RunWith(MockitoJUnit44Runner.class)
public class FileReadingMessageSourceTests {
private FileReadingMessageSource source;
- private File inputDirectoryMock = createMock(File.class);
+ @Mock
+ private File inputDirectoryMock;
- private Resource inputDirectoryResourceMock = createNiceMock(Resource.class);
+ @Mock
+ private Resource inputDirectoryResourceMock;
- private File fileMock = createMock(File.class);
+ @Mock
+ private File fileMock;
- private Comparator comparator = createMock(Comparator.class);
-
- private Object[] allMocks = new Object[] { inputDirectoryMock, fileMock, inputDirectoryResourceMock, comparator };
+ @Mock
+ private Comparator comparator;
public void prepResource() throws Exception {
- expect(inputDirectoryResourceMock.exists()).andReturn(true).anyTimes();
- expect(inputDirectoryResourceMock.getFile()).andReturn(inputDirectoryMock).anyTimes();
- expect(inputDirectoryMock.canRead()).andReturn(true);
- replay(inputDirectoryResourceMock, inputDirectoryMock);
+ when(inputDirectoryResourceMock.exists()).thenReturn(true);
+ when(inputDirectoryResourceMock.getFile()).thenReturn(inputDirectoryMock);
+ when(inputDirectoryMock.canRead()).thenReturn(true);
}
@Before
@@ -59,59 +64,64 @@ public class FileReadingMessageSourceTests {
prepResource();
this.source = new FileReadingMessageSource(comparator);
source.setInputDirectory(inputDirectoryResourceMock);
- reset(allMocks);
}
@Test
public void straightProcess() throws Exception {
- expect(inputDirectoryMock.listFiles()).andReturn(new File[] { fileMock });
- replay(allMocks);
+ when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock });
source.onSend(source.receive());
- verify(allMocks);
}
@Test
public void requeueOnFailure() throws Exception {
- expect(inputDirectoryMock.listFiles()).andReturn(new File[] { fileMock }).times(2);
- replay(allMocks);
+ when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock });
Message received = source.receive();
assertNotNull(received);
source.onFailure(received, new RuntimeException("failed"));
assertEquals(received.getPayload(), source.receive().getPayload());
- verify(allMocks);
+ verify(inputDirectoryMock,times(1)).listFiles();
+ }
+
+ @Test
+ public void scanEachPoll() throws Exception {
+ File anotherFileMock = mock(File.class);
+ when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock, anotherFileMock });
+ source.setScanEachPoll(true);
+ assertNotNull(source.receive());
+ assertNotNull(source.receive());
+ assertNull(source.receive());
+ verify(inputDirectoryMock,times(3)).listFiles();
}
@Test
public void noDuplication() throws Exception {
- expect(inputDirectoryMock.listFiles()).andReturn(new File[] { fileMock }).times(2);
- replay(allMocks);
+ when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock });
Message received = source.receive();
assertNotNull(received);
assertEquals(fileMock, received.getPayload());
assertNull(source.receive());
- verify(allMocks);
+ verify(inputDirectoryMock,times(2)).listFiles();
}
@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);
+ File file1 = mock(File.class);
+ File file2 = mock(File.class);
+ File file3 = mock(File.class);
+
+ // record the comparator to reverse order the files
+ when(comparator.compare(file1, file2)).thenReturn(1);
+ when(comparator.compare(file1, file3)).thenReturn(1);
+ when(comparator.compare(file2, file3)).thenReturn(1);
+ when(comparator.compare(file2, file1)).thenReturn(-1);
+ when(comparator.compare(file3, file1)).thenReturn(-1);
+ when(comparator.compare(file3, file2)).thenReturn(-1);
+
+ when(inputDirectoryMock.listFiles()).thenReturn(new File[] { file2, file3, file1 });
assertSame(file3, source.receive().getPayload());
assertSame(file2, source.receive().getPayload());
assertSame(file1, source.receive().getPayload());
assertNull(source.receive());
- verify(allMocks);
+ verify(inputDirectoryMock,times(2)).listFiles();
}
}