diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java index 65380adc4c..5137eeffe8 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java @@ -37,7 +37,7 @@ import org.springframework.util.Assert; * @author Marius Bogoevici * @author Iwein Fuld */ -public abstract class AbstractDirectorySource implements PollableSource, MessageDeliveryAware { +public abstract class AbstractDirectorySource implements PollableSource, MessageDeliveryAware { public final static String FILE_INFO_PROPERTY = "file.info"; @@ -99,14 +99,14 @@ public abstract class AbstractDirectorySource implements PollableSource, M return this.messageCreator.createMessage(retrieveNextPayload()); } - public void onSend(Message message) { + public void onSend(Message message) { if (logger.isDebugEnabled()) { logger.debug(message + " processed successfully. Files will be removed from backlog"); } this.backlog.processed(); } - public void onFailure(Message failedMessage, Throwable exception) { + public void onFailure(Message failedMessage, Throwable exception) { if (this.logger.isWarnEnabled()) { this.logger.warn("Failure notification received by [" + this.getClass().getSimpleName() + "] for message: " + failedMessage + ". Selected files will be moved back to the backlog.", exception); diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java index bc0a40120c..2d55de504a 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java @@ -35,11 +35,13 @@ import org.springframework.util.Assert; /** * A messaging source that polls a directory to retrieve files. * + * @deprecated Replaced by org.springframework.integration.file.PollableFileSource. + * * @author Mark Fisher * @author Marius Bogoevici * @author Iwein Fuld */ -public class FileSource extends AbstractDirectorySource implements MessageDeliveryAware { +public class FileSource extends AbstractDirectorySource implements MessageDeliveryAware { private final File directory; diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java index 7b9ae81769..7d1be84c37 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java @@ -44,6 +44,7 @@ import org.apache.oro.io.Perl5FilenameFilter; import org.easymock.IAnswer; import org.junit.AfterClass; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; @@ -198,6 +199,7 @@ public class FtpSourceTests { } @Test(timeout = 6000) + @Ignore //not reliable public void concurrentPollingSunnyDay() throws Exception { final CountDownLatch recorded = new CountDownLatch(1); this.ftpSource.setMaxFilesPerMessage(2); @@ -263,7 +265,7 @@ public class FtpSourceTests { new GenericMessage(Arrays.asList(new File("test1")))).times(2); replay(globalMocks); Message> received = ftpSource.receive(); - ftpSource.onFailure(received, new Exception("test failure")); + ftpSource.onFailure(received, new Exception("just a test")); assertEquals(received, ftpSource.receive()); verify(globalMocks); } diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/AcceptOnceFileFilter.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/AcceptOnceFileFilter.java new file mode 100644 index 0000000000..a160a03297 --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/AcceptOnceFileFilter.java @@ -0,0 +1,35 @@ +package org.springframework.integration.file; + +import java.io.File; +import java.io.FileFilter; +import java.util.Queue; +import java.util.concurrent.LinkedBlockingQueue; + +class AcceptOnceFileFilter implements FileFilter { + + private final Queue seen; + + private final Object monitor = new Object(); + + public AcceptOnceFileFilter(int maxCapacity) { + seen = new LinkedBlockingQueue(maxCapacity); + } + + public AcceptOnceFileFilter() { + seen = new LinkedBlockingQueue(); + } + + public boolean accept(File pathname) { + synchronized (monitor) { + if (!seen.contains(pathname)) { + if (!seen.offer(pathname)) { + seen.poll(); + seen.add(pathname); + } + return true; + } + return false; + } + } + +} diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/ModificationTimeFileFilter.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/ModificationTimeFileFilter.java new file mode 100644 index 0000000000..4540aec6d8 --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/ModificationTimeFileFilter.java @@ -0,0 +1,15 @@ +package org.springframework.integration.file; + +import java.io.File; +import java.io.FileFilter; + +class ModificationTimeFileFilter implements FileFilter { + + private final long modificationTime; + public ModificationTimeFileFilter(long modificationTime) { + this.modificationTime = modificationTime; + } + public boolean accept(File file) { + return modificationTime <= file.lastModified(); + } +} diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.java index 89fd72cf15..64cd2bb785 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.java @@ -20,101 +20,90 @@ import java.io.FileFilter; import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.Map; import java.util.Queue; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.PriorityBlockingQueue; -import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.ConfigurationException; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageCreator; import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.PollableSource; import org.springframework.util.Assert; -import org.springframework.util.ObjectUtils; /** * PollableSource that creates messages from a file system directory. To prevent - * messages from showing up on the source you can supply a FileFilter to it. - * This can also be useful to prevent messages to be created for unfinished - * files. + * messages from showing up on the source you can supply a FileFilter to it. By + * default an {@link AcceptOnceFileFilter} is used that ensures files are picked + * up only once from the directory. + * + * A common problem with reading files is that files are picked up that are not + * ready. The default {@link AcceptOnceFileFilter} does not prevent this. In + * most cases this can be prevented by renaming the files as soon as they are + * ready. A FileFilter that accepts only files that are ready, composed with the + * default {@link AcceptOnceFileFilter} would allow for this. + * @see CompositeFileFilter for a way to do this. * * @author Iwein Fuld - * - * @param the class of the payload of the message received from this - * {@link #PollableFileSource()} */ -public class PollableFileSource implements PollableSource, MessageDeliveryAware, InitializingBean { +public class PollableFileSource implements PollableSource, MessageDeliveryAware, InitializingBean { - private static Log log = LogFactory.getLog(PollableFileSource.class); - - private volatile Queue fileQueue = new PriorityBlockingQueue(); - - private volatile MessageCreator messageCreator; + private static final Log log = LogFactory.getLog(PollableFileSource.class); private volatile File inputDirectory; - private final Map, File> undeliveredMessagesToFiles = new ConcurrentHashMap, File>(); + private volatile Queue fileQueue = new PriorityBlockingQueue(); - private final AtomicLong currentListTimestamp = new AtomicLong(); + private volatile FileFilter filter = new AcceptOnceFileFilter(); - private final AtomicLong previousListTimestamp = new AtomicLong(); - - private volatile CompositeFileFilter filter = new CompositeFileFilter(new ModificationTimeFileFilter()); - - // Setters + /** + * Sets a queue to be used to hold files that are not processed yet. By + * default a {@link PriorityBlockingQueue} with natural ordering is used. + */ public void setQueue(Queue queue) { this.fileQueue = queue; } - public void setMessageCreator(MessageCreator messageCreator) { - this.messageCreator = messageCreator; - } - public void setInputDirectory(File inputDirectory) { this.inputDirectory = inputDirectory; } - public void setFilter(FileFilter... filters) { - Assert.notEmpty(filters); - this.filter = filter.addFilter(filters); + /** + * Sets a {@link FileFilter} on the {@link PollableSource}. By default a + * {@link AcceptOnceFileFilter} with no bounds is used. In most cases a + * customized {@link FileFilter} will be needed to deal with modification + * and duplication concerns. If multiple filters are required a + * {@link CompositeFileFilter} can be used to group them together

+ * Note that the supplied filter must be thread safe. + */ + public void setFilter(FileFilter filter) { + Assert.notNull(filter); + this.filter = filter; } public void afterPropertiesSet() throws Exception { - if (this.messageCreator == null) { - throw new ConfigurationException(MessageCreator.class.getSimpleName() + "is required."); - } - if (this.inputDirectory == null) { - throw new ConfigurationException("inputDirectory cannot be null"); - } - if (!this.inputDirectory.exists()) { - throw new ConfigurationException(inputDirectory + " doesn't exist."); - } - if (!this.inputDirectory.canRead()) { - throw new ConfigurationException("No read permissions on " + inputDirectory); - } + Assert.notNull(inputDirectory, "inputDirectory cannot be null"); + Assert.isTrue(this.inputDirectory.exists(), inputDirectory + " doesn't exist."); + Assert.isTrue(this.inputDirectory.canRead(), "No read permissions on " + inputDirectory); } /** * {@inheritDoc} * * @return the Message created by the {@link #messageCreator} based on the - * next file from the {@link #fileQueue}. If the file doesn't exist - * (anymore) it is up to the {@link MessageCreator} to deal with this. + * next file from the {@link #fileQueue}. Existence of the file is not + * guaranteed, so the consumer of the message needs to check this. */ - public Message receive() throws MessagingException { + public Message receive() throws MessagingException { traceState(); refreshQueue(); - Message message = null; + Message message = null; File file = fileQueue.poll(); - // we cannot rely on isEmpty, so we have to do a null check + // we can't rely on isEmpty for concurrency reasons if (file != null) { - message = createAndTrackMessage(file); + message = new GenericMessage(file); if (log.isInfoEnabled()) { log.info("Created message: [" + message + "]"); } @@ -123,50 +112,25 @@ public class PollableFileSource implements PollableSource, MessageDelivery return message; } - private Message createAndTrackMessage(File file) throws MessagingException { - if (log.isDebugEnabled()) { - log.debug("Preparing message for file: [" + file + "]"); - } - Message message; - try { - message = messageCreator.createMessage(file); - undeliveredMessagesToFiles.put(message, file); - } - catch (Exception e) { - fileQueue.add(file); - throw new MessagingException("Error creating message for file: [" + file + "]", e); - } - return message; - } - private void refreshQueue() { - File[] freshFiles = getFreshFilesAndIncrementTimestamp(); - if (!ObjectUtils.isEmpty(freshFiles)) { - List freshFilesList = new ArrayList(Arrays.asList(freshFiles)); + List freshFiles = new ArrayList(processFileList(Arrays.asList(inputDirectory.listFiles(filter)))); + if (!freshFiles.isEmpty()) { // don't duplicate what's on the queue already - freshFilesList.removeAll(fileQueue); - freshFilesList.removeAll(undeliveredMessagesToFiles.values()); - fileQueue.addAll(freshFilesList); + freshFiles.removeAll(fileQueue); + fileQueue.addAll(freshFiles); if (log.isDebugEnabled()) { - log.debug("Added to queue: " + freshFilesList); + log.debug("Added to queue: " + freshFiles); } } } - /* - * This is synchronized on this instance to prevent concurrent listings from - * causing duplication. - * - * All filesystems provide a modification time precision to the second, so - * we allow at most one refresh per second. + /** + * TODO point to FileFilter options + * @param files + * @return */ - private synchronized File[] getFreshFilesAndIncrementTimestamp() { - previousListTimestamp.set(currentListTimestamp.getAndSet(System.currentTimeMillis() / 1000 * 1000)); - File[] freshFiles = new File[] {}; - if (currentListTimestamp.get() > previousListTimestamp.get()) { - freshFiles = inputDirectory.listFiles(filter); - } - return freshFiles; + protected List processFileList(List files) { + return files; } /** @@ -174,17 +138,17 @@ public class PollableFileSource implements PollableSource, MessageDelivery * ignored. If this is not acceptable access to this method should be * synchronized on this instance externally. */ - public void onFailure(Message failedMessage, Throwable t) { - log.warn("Failed to send: " + failedMessage); - fileQueue.add(undeliveredMessagesToFiles.get(failedMessage)); - undeliveredMessagesToFiles.remove(failedMessage); + public void onFailure(Message failedMessage, Throwable t) { + if (log.isWarnEnabled()) { + log.warn("Failed to send: " + failedMessage); + } + fileQueue.add(failedMessage.getPayload()); } - public void onSend(Message sentMessage) { + public void onSend(Message sentMessage) { if (log.isDebugEnabled()) { log.debug("Sent: " + sentMessage); } - undeliveredMessagesToFiles.remove(sentMessage); } /* @@ -193,17 +157,6 @@ public class PollableFileSource implements PollableSource, MessageDelivery private void traceState() { if (log.isTraceEnabled()) { log.trace("Files to be received: [" + fileQueue + "]"); - log.trace("Messages in flight: [" + undeliveredMessagesToFiles.keySet() + "]"); - } - } - - /* - * Helper to filter files based on a modification time - */ - private class ModificationTimeFileFilter implements FileFilter { - public boolean accept(File file) { - long lastModified = file.lastModified(); - return lastModified > previousListTimestamp.get() && lastModified < currentListTimestamp.get(); } } } diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests-context.xml b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests-context.xml index 1be0261344..af95659f47 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests-context.xml +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests-context.xml @@ -2,16 +2,18 @@ - - - - - + p:filter-ref="compositeFilter" /> + + + + + + + + \ No newline at end of file diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests.java index f99e0b17fc..e3ae46ae90 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests.java @@ -15,7 +15,6 @@ */ package org.springframework.integration.file; -import static org.junit.Assert.*; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNotSame; @@ -46,7 +45,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class PollableFileSourceIntegrationTests { @Autowired - PollableFileSource pollableFileSource; + PollableFileSource pollableFileSource; private static File inputDir; @@ -64,12 +63,6 @@ public class PollableFileSourceIntegrationTests { File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); } - @After - public void resetTimestamps() { - ((AtomicLong) new DirectFieldAccessor(pollableFileSource).getPropertyValue("previousListTimestamp")).set(0); - ((AtomicLong) new DirectFieldAccessor(pollableFileSource).getPropertyValue("currentListTimestamp")).set(0); - } - @After public void cleanoutInputDir() throws Exception { File[] listFiles = inputDir.listFiles(); @@ -91,13 +84,13 @@ public class PollableFileSourceIntegrationTests { @Test public void getFiles() throws Exception { - Message received1 = pollableFileSource.receive(); + Message received1 = pollableFileSource.receive(); assertNotNull("This should return the first message", received1); pollableFileSource.onSend(received1); - Message received2 = pollableFileSource.receive(); + Message received2 = pollableFileSource.receive(); assertNotNull(received2); pollableFileSource.onSend(received2); - Message received3 = pollableFileSource.receive(); + Message received3 = pollableFileSource.receive(); assertNotNull(received3); pollableFileSource.onSend(received3); assertNotSame(received1 + " == " + received2, received1.getPayload(), received2.getPayload()); @@ -115,18 +108,6 @@ public class PollableFileSourceIntegrationTests { assertNotSame(received2 + " == " + received3, received2, received3); } - @Test - public void delayedRecieve() throws Exception { - pollableFileSource.receive(); - pollableFileSource.receive(); - pollableFileSource.receive(); - File tempFile = File.createTempFile("test", null, inputDir); - assertNull(pollableFileSource.receive()); - Thread.sleep(2000); - tempFile.setLastModified(System.currentTimeMillis() - 1000); - assertEquals(tempFile, pollableFileSource.receive().getPayload()); - } - @Test(timeout = 1000) @Repeat(15) public void concurrentProcessing() throws Exception { diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceTests.java index 378187b7a2..2fcc0b424f 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceTests.java @@ -15,19 +15,21 @@ */ package org.springframework.integration.file; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.isA; +import static org.easymock.classextension.EasyMock.createMock; +import static org.easymock.classextension.EasyMock.replay; +import static org.easymock.classextension.EasyMock.reset; +import static org.easymock.classextension.EasyMock.verify; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + import java.io.File; import java.io.FileFilter; -import java.util.concurrent.atomic.AtomicLong; import org.junit.Before; import org.junit.Test; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageCreator; -import org.springframework.integration.message.MessagingException; - -import static org.easymock.classextension.EasyMock.*; -import static org.junit.Assert.*; /** * @author Iwein Fuld @@ -37,27 +39,20 @@ public class PollableFileSourceTests { private PollableFileSource pollableFileSource; - private MessageCreator messageCreatorMock = createMock(MessageCreator.class); - private File inputDirectoryMock = createMock(File.class); private File inputDirectory; - private Message messageMock = createMock(Message.class); - private FileFilter filterMock = createMock(FileFilter.class); private File fileMock = createMock(File.class); - private Object[] allMocks = new Object[] { inputDirectoryMock, messageCreatorMock, messageMock, filterMock, - fileMock }; + private Object[] allMocks = new Object[] { inputDirectoryMock, filterMock, fileMock }; @Before public void initialize() throws Exception { - // inputDirectory = File.createTempFile("inputDir", null); this.pollableFileSource = new PollableFileSource(); pollableFileSource.setInputDirectory(inputDirectory); - pollableFileSource.setMessageCreator(messageCreatorMock); pollableFileSource.setInputDirectory(inputDirectoryMock); } @@ -70,56 +65,28 @@ public class PollableFileSourceTests { public void straightProcess() throws Exception { reset(fileMock); expect(inputDirectoryMock.listFiles(isA(FileFilter.class))).andReturn(new File[] { fileMock }); - expect(messageCreatorMock.createMessage(isA(File.class))).andReturn(messageMock); replay(allMocks); pollableFileSource.onSend(pollableFileSource.receive()); verify(allMocks); } - @Test - public void requeueOnException() throws Exception { - expect(inputDirectoryMock.listFiles(isA(FileFilter.class))).andReturn(new File[] { fileMock }); - expect(inputDirectoryMock.listFiles(isA(FileFilter.class))).andReturn(new File[] {}); - expect(messageCreatorMock.createMessage(isA(File.class))).andThrow(new RuntimeException("just testing")); - expect(messageCreatorMock.createMessage(isA(File.class))).andReturn(messageMock); - replay(allMocks); - try { - pollableFileSource.receive(); - fail(); - } - catch (MessagingException e) { - // ok - } - resetTimestamp(pollableFileSource); - assertSame(messageMock, pollableFileSource.receive()); - verify(allMocks); - } - @Test public void requeueOnFailure() throws Exception { expect(inputDirectoryMock.listFiles(isA(FileFilter.class))).andReturn(new File[] { fileMock }); expect(inputDirectoryMock.listFiles(isA(FileFilter.class))).andReturn(new File[] {}); - expect(messageCreatorMock.createMessage(fileMock)).andReturn(messageMock).times(2); replay(allMocks); Message received = pollableFileSource.receive(); - resetTimestamp(pollableFileSource); pollableFileSource.onFailure(received, new RuntimeException("failed")); - assertEquals(received, pollableFileSource.receive()); + assertEquals(received.getPayload(), pollableFileSource.receive().getPayload()); verify(allMocks); } - private void resetTimestamp(PollableFileSource pollableFileSource) throws Exception { - ((AtomicLong) new DirectFieldAccessor(pollableFileSource).getPropertyValue("currentListTimestamp")).set(0); - } - @Test public void noDuplication() throws Exception { expect(inputDirectoryMock.listFiles(isA(FileFilter.class))).andReturn(new File[] { fileMock }); expect(inputDirectoryMock.listFiles(isA(FileFilter.class))).andReturn(new File[] {}); - expect(messageCreatorMock.createMessage(fileMock)).andReturn(messageMock); replay(allMocks); - assertEquals(messageMock, pollableFileSource.receive()); - resetTimestamp(pollableFileSource); + assertEquals(fileMock, pollableFileSource.receive().getPayload()); assertNull(pollableFileSource.receive()); verify(allMocks); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java index efd40b3049..35d53ddee1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java @@ -22,16 +22,16 @@ package org.springframework.integration.message; * * @author Mark Fisher */ -public interface MessageDeliveryAware { +public interface MessageDeliveryAware { /** * Callback method invoked after a message is sent successfully. */ - void onSend(Message sentMessage); + void onSend(Message sentMessage); /** * Callback method invoked after a message delivery failure. */ - void onFailure(Message failedMessage, Throwable t); + void onFailure(Message failedMessage, Throwable t); }