Complete - task 6: Create PollableFileSource

Last review done by Mark and Arjen, modification time concern pushed out.
Added parameter to MessageDeliveryAware

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<File> seen;
+
+	private final Object monitor = new Object();
+
+	public AcceptOnceFileFilter(int maxCapacity) {
+		seen = new LinkedBlockingQueue<File>(maxCapacity);
+	}
+
+	public AcceptOnceFileFilter() {
+		seen = new LinkedBlockingQueue<File>();
+	}
+
+	public boolean accept(File pathname) {
+		synchronized (monitor) {
+			if (!seen.contains(pathname)) {
+				if (!seen.offer(pathname)) {
+					seen.poll();
+					seen.add(pathname);
+				}
+				return true;
+			}
+			return false;
+		}
+	}
+
+}

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();
+	}
+}

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 <T> the class of the payload of the message received from this
- * {@link #PollableFileSource()}
  */
-public class PollableFileSource<T> implements PollableSource<T>, MessageDeliveryAware, InitializingBean {
+public class PollableFileSource implements PollableSource<File>, MessageDeliveryAware<File>, InitializingBean {
 
-	private static Log log = LogFactory.getLog(PollableFileSource.class);
-
-	private volatile Queue<File> fileQueue = new PriorityBlockingQueue<File>();
-
-	private volatile MessageCreator<File, T> messageCreator;
+	private static final Log log = LogFactory.getLog(PollableFileSource.class);
 
 	private volatile File inputDirectory;
 
-	private final Map<Message<T>, File> undeliveredMessagesToFiles = new ConcurrentHashMap<Message<T>, File>();
+	private volatile Queue<File> fileQueue = new PriorityBlockingQueue<File>();
 
-	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<File> queue) {
 		this.fileQueue = queue;
 	}
 
-	public void setMessageCreator(MessageCreator<File, T> 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 <p/>
+	 * <b>Note that the supplied filter must be thread safe</b>.
+	 */
+	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<T> receive() throws MessagingException {
+	public Message<File> receive() throws MessagingException {
 		traceState();
 		refreshQueue();
-		Message<T> message = null;
+		Message<File> 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>(file);
 			if (log.isInfoEnabled()) {
 				log.info("Created message: [" + message + "]");
 			}
@@ -123,50 +112,25 @@ public class PollableFileSource<T> implements PollableSource<T>, MessageDelivery
 		return message;
 	}
 
-	private Message<T> createAndTrackMessage(File file) throws MessagingException {
-		if (log.isDebugEnabled()) {
-			log.debug("Preparing message for file: [" + file + "]");
-		}
-		Message<T> 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<File> freshFilesList = new ArrayList<File>(Arrays.asList(freshFiles));
+		List<File> freshFiles = new ArrayList<File>(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<File> processFileList(List<File> files) {
+		return files;
 	}
 
 	/**
@@ -174,17 +138,17 @@ public class PollableFileSource<T> implements PollableSource<T>, 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<File> 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<File> sentMessage) {
 		if (log.isDebugEnabled()) {
 			log.debug("Sent: " + sentMessage);
 		}
-		undeliveredMessagesToFiles.remove(sentMessage);
 	}
 
 	/*
@@ -193,17 +157,6 @@ public class PollableFileSource<T> implements PollableSource<T>, 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();
 		}
 	}
 }

org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests-context.xml

@@ -2,16 +2,18 @@
 <beans xmlns="http://www.springframework.org/schema/beans"
 	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p"
 	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
-
-	<bean id="pollableFileSource"
-		class="org.springframework.integration.file.PollableFileSource"
+	<bean id="pollableFileSource" class="org.springframework.integration.file.PollableFileSource"
 		p:inputDirectory="${java.io.tmpdir}/PollableFileSourceIntegrationTests"
-		p:messageCreator-ref="messageCreator" p:filter-ref="someFilter" />
-		
-	<bean id="someFilter"
-		class="org.springframework.integration.file.TestFileFilter"/>
-	<bean id="messageCreator"
-		class="org.springframework.integration.message.DefaultMessageCreator" />
+		p:filter-ref="compositeFilter" />
+	<bean id="compositeFilter"
+		class="org.springframework.integration.file.CompositeFileFilter">
+		<constructor-arg>
+			<list>
+				<bean class="org.springframework.integration.file.AcceptOnceFileFilter" />
+				<bean class="org.springframework.integration.file.TestFileFilter" />
+			</list>
+		</constructor-arg>
+	</bean>
 	<bean
 		class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
 </beans>

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<File> 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<File> received1 = pollableFileSource.receive();
 		assertNotNull("This should return the first message", received1);
 		pollableFileSource.onSend(received1);
-		Message<?> received2 = pollableFileSource.receive();
+		Message<File> received2 = pollableFileSource.receive();
 		assertNotNull(received2);
 		pollableFileSource.onSend(received2);
-		Message<?> received3 = pollableFileSource.receive();
+		Message<File> 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 {

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