From a0da740a3d5096a75ff1c65ab11f6badda531519 Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Mon, 23 Feb 2009 16:19:36 +0000 Subject: [PATCH] OPEN - issue INT-583: FileReadingMessageSource rescanning directory every pass http://jira.springframework.org/browse/INT-583 Fixed issue, added flag to go back to old behavior. Refactored test to Mockito. --- .../.classpath | 2 + org.springframework.integration.file/ivy.xml | 1 + .../file/FileReadingMessageSource.java | 28 +++++-- .../file/FileReadingMessageSourceTests.java | 82 +++++++++++-------- 4 files changed, 71 insertions(+), 42 deletions(-) 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(); } }