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

IN PROGRESS - issue INT-541: add optional Comparator argument to FileReadingMessageSource (to pass into the PriorityBlockingQueue constructor)

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

Added constructor argument and test.
This commit is contained in:
Iwein Fuld
2009-01-09 14:49:14 +00:00
parent 9d1403463b
commit 747050ab8c
2 changed files with 64 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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.
* <p/>
* 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.
* <p/>
* 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.
* <p/>
* 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<File> {
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<File> {
* {@link ConcurrentModificationException} in Java 5. There is no locking
* around the queue, so there is also no iteration.
*/
private final Queue<File> toBeReceived = new PriorityBlockingQueue<File>();
private final Queue<File> toBeReceived;
private volatile FileListFilter filter = new AcceptOnceFileListFilter();
/**
* Creates a FileReadingMessageSource with a naturally ordered queue.
*/
public FileReadingMessageSource() {
toBeReceived = new PriorityBlockingQueue<File>(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<File> receptionOrderComparator) {
toBeReceived = new PriorityBlockingQueue<File>(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<File> {
* and duplication concerns. If multiple filters are required a
* {@link CompositeFileListFilter} can be used to group them together
* <p/>
* <b>Note that the supplied filter must be thread safe</b>.
* <b>Note that the supplied filter must be thread safe.</b>.
*/
public void setFilter(FileListFilter filter) {
Assert.notNull(filter, "'filter' should not be null");

39
org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java
View File

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