Incomplete: Create PollableFileSource (this is the first step to make Backlog obsolete)

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/CompositeFileFilter.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.adapter.file.replacement;
+
+import java.io.File;
+import java.io.FileFilter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Composition that delegates to multiple {@link FileFilter}s. The compostition
+ * is AND based, meaning that all filters must {@link #accept(File)} in order
+ * for a file to be accepted by the composite.
+ * 
+ * @author Iwein Fuld
+ */
+public class CompositeFileFilter implements FileFilter {
+
+	private final List<FileFilter> fileFilters;
+
+	public CompositeFileFilter(FileFilter... fileFilters) {
+		this.fileFilters = new ArrayList<FileFilter>(Arrays.asList(fileFilters));
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	public boolean accept(File pathname) {
+		for (FileFilter fileFilter : fileFilters) {
+			if (!fileFilter.accept(pathname)) {
+				return false;
+			}
+		}
+		return true;
+	}
+
+	public void addFilter(FileFilter... filters) {
+		fileFilters.addAll(Arrays.asList(filters));
+	}
+
+	public static CompositeFileFilter with(FileFilter... fileFilters) {
+		return new CompositeFileFilter(fileFilters);
+	}
+
+}

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/ModificationTimeFileFilter.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.adapter.file.replacement;
+
+import java.io.File;
+import java.io.FileFilter;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.springframework.util.Assert;
+
+/**
+ * FileFilter that passes all files with a lastModified after a specified
+ * timestamp. The filter uses a mutable AtomicLong so the timestamp can be
+ * changed during the lifetime of the filter.
+ * 
+ * @author Iwein Fuld
+ * 
+ */
+public class ModificationTimeFileFilter implements FileFilter {
+
+	private final AtomicLong minimumLastModified;
+
+	public ModificationTimeFileFilter(AtomicLong lastRecieveTimestamp) {
+		this.minimumLastModified = lastRecieveTimestamp;
+	}
+
+	public boolean accept(File file) {
+		Assert.notNull(file);
+		return file.exists() && file.lastModified() > minimumLastModified.get();
+	}
+
+}

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/PollableFileSource.java

@@ -0,0 +1,181 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.adapter.file.replacement;
+
+import java.io.File;
+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.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;
+
+/**
+ * PollableSource that creates messages from a file system directory.
+ * 
+ * @author Iwein Fuld
+ * 
+ * @param <T>
+ */
+public class PollableFileSource<T> implements PollableSource<T>, MessageDeliveryAware, InitializingBean {
+
+	private Log log = LogFactory.getLog(this.getClass());
+
+	private Queue<File> queue = new PriorityBlockingQueue<File>();
+
+	private MessageCreator<File, T> messageCreator;
+
+	private File inputDirectory;
+
+	private final Map<Message<T>, File> tracker = new ConcurrentHashMap<Message<T>, File>();
+
+	private final AtomicLong lastListTimestamp = new AtomicLong();
+
+	private final CompositeFileFilter filter = CompositeFileFilter.with(new ModificationTimeFileFilter(
+			lastListTimestamp));
+
+	public Message<T> receive() throws MessagingException {
+		refreshQueue(queue);
+		File file = queue.poll();
+		// ignore files that have been deleted
+		while (file != null && !file.exists()) {
+			file = queue.poll();
+		}
+		if (file != null) {
+			if (log.isInfoEnabled()) {
+				log.info("preparing to send a message for file: [" + file + "]");
+			}
+			return createAndTrackMessage(file);
+		}
+		// queue is empty
+		return null;
+	}
+
+	private Message<T> createAndTrackMessage(File file) throws MessagingException {
+		Message<T> message;
+		try {
+			message = messageCreator.createMessage(file);
+			tracker.put(message, file);
+		}
+		catch (Exception e) {
+			log.warn("Error occured while attempting to create message. Requeuing file [" + file + "]", e);
+			queue.add(file);
+			throw new MessagingException("Error while polling for messages", e);
+		}
+		if (log.isDebugEnabled()) {
+			log.debug("created message: [" + message + "]");
+		}
+		if (log.isTraceEnabled()) {
+			log.trace("queue: [" + queue + "]");
+			log.trace("tracker: [" + tracker.values() + "]");
+		}
+		return message;
+	}
+
+	private void refreshQueue(Queue<File> queue) {
+		File[] freshFiles = getFreshFilesAndIncrementTimestamp();
+		if (freshFiles != null) {
+			List<File> freshFilesList = new ArrayList<File>(Arrays.asList(freshFiles));
+			// don't duplicate what's on the queue already
+			freshFilesList.removeAll(queue);
+			freshFilesList.removeAll(tracker.values());
+			if (log.isTraceEnabled()) {
+				log.trace("queue: [" + queue + "]");
+				log.trace("tracker: [" + tracker.values() + "]");
+			}
+			queue.addAll(freshFilesList);
+			if (log.isDebugEnabled()) {
+				log.debug("added to queue: " + freshFilesList);
+			}
+		}
+	}
+
+	/*
+	 * This is synchronized to prevent concurrent listings from causing
+	 * duplication.
+	 */
+	private synchronized File[] getFreshFilesAndIncrementTimestamp() {
+		File[] freshFiles = inputDirectory.listFiles(filter);
+		lastListTimestamp.set(System.currentTimeMillis());
+		return freshFiles;
+	}
+
+	/**
+	 * In concurrent scenarios onFailure() might cause failing files to be
+	 * ignored. If this is not acceptable access to this method should be
+	 * synchronized.
+	 * 
+	 * {@inheritDoc} 
+	 */
+	public void onFailure(Message<?> failedMessage, Throwable t) {
+		log.warn("not sent: " + failedMessage);
+		queue.add(tracker.get(failedMessage));
+		tracker.remove(failedMessage);
+	}
+
+	public void onSend(Message<?> sentMessage) {
+		if (log.isDebugEnabled()) {
+			log.debug("sent: " + sentMessage);
+		}
+		tracker.remove(sentMessage);
+	}
+
+	// Setters
+	public void setQueue(Queue<File> queue) {
+		this.queue = queue;
+	}
+
+	public void setMessageCreator(MessageCreator<File, T> messageCreator) {
+		this.messageCreator = messageCreator;
+	}
+
+	public void setInputDirectory(File inputDirectory) {
+		this.inputDirectory = inputDirectory;
+	}
+
+	public void setFilter(FileFilter filter) {
+		this.filter.addFilter(filter);
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		if (this.messageCreator == null) {
+			throw new ConfigurationException("I need a " + MessageCreator.class.getSimpleName()
+					+ " in order to function.");
+		}
+		if (this.inputDirectory == null) {
+			throw new ConfigurationException("I need an inputDirectory to read from.");
+		}
+		if (!this.inputDirectory.exists()) {
+			throw new ConfigurationException(inputDirectory + " doesn't exist.");
+		}
+		if (!this.inputDirectory.canRead()) {
+			throw new ConfigurationException("I can't read from " + inputDirectory);
+		}
+	}
+}

org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/CompositeFileFilterTest.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.adapter.file.replacement;
+
+import java.io.File;
+import java.io.FileFilter;
+import static org.easymock.classextension.EasyMock.*;
+import static org.junit.Assert.*;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * @author Iwein Fuld
+ */
+public class CompositeFileFilterTest {
+
+	private FileFilter fileFilterMock = createMock(FileFilter.class);
+
+	private CompositeFileFilter compositeFileFilter;
+
+	private File fileMock = createNiceMock(File.class);
+
+	@Before
+	public void initialize() {
+		compositeFileFilter = CompositeFileFilter.with(fileFilterMock, fileFilterMock);
+	}
+
+	@Test
+	public void forwardedToFilters() throws Exception {
+		expect(fileFilterMock.accept(fileMock)).andReturn(true).times(2);
+		replay(fileFilterMock);
+		assertTrue(compositeFileFilter.accept(fileMock));
+		verify(fileFilterMock);
+	}
+
+	@Test
+	public void forwardedToAddedFilters() throws Exception {
+		expect(fileFilterMock.accept(fileMock)).andReturn(true).times(4);
+		replay(fileFilterMock);
+		compositeFileFilter.addFilter(fileFilterMock, fileFilterMock);
+		assertTrue(compositeFileFilter.accept(fileMock));
+		verify(fileFilterMock);
+	}
+
+	@Test
+	public void notForwardedWhenNegative() throws Exception {
+		expect(fileFilterMock.accept(fileMock)).andReturn(true).times(2);
+		expect(fileFilterMock.accept(fileMock)).andReturn(false).times(1);
+		replay(fileFilterMock);
+		compositeFileFilter.addFilter(fileFilterMock, fileFilterMock, fileFilterMock, fileFilterMock);
+		assertFalse(compositeFileFilter.accept(fileMock));
+		verify(fileFilterMock);
+	}
+}

org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/ModificationTimeFileFilterTests.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.adapter.file.replacement;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * @author Iwein Fuld
+ */
+public class ModificationTimeFileFilterTests {
+
+	private AtomicLong timestamp = new AtomicLong(System.currentTimeMillis() - 5000);
+
+	private ModificationTimeFileFilter modificationTimeFileFilter;
+
+	@Before
+	public void initialize() {
+		modificationTimeFileFilter = new ModificationTimeFileFilter(timestamp);
+	}
+
+	@Test
+	public void accept() throws Exception {
+		File tempFile = File.createTempFile("test", null);
+		assertTrue("File modification date too early", tempFile.lastModified() > (timestamp.get()));
+		assertTrue("not accepted", modificationTimeFileFilter.accept(tempFile));
+	}
+}

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

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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.adapter.file.replacement.PollableFileSource"
+		p:inputDirectory="${java.io.tmpdir}/PollableFileSourceIntegrationTests" 
+		p:messageCreator-ref="messageCreator"/>
+
+	<bean id="messageCreator"
+		class="org.springframework.integration.message.DefaultMessageCreator" />
+	<bean
+		class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
+</beans>

org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceIntegrationTests.java

@@ -0,0 +1,183 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.adapter.file.replacement;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertNull;
+
+import java.io.File;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.message.Message;
+import org.springframework.test.annotation.Repeat;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Iwein Fuld
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+public class PollableFileSourceIntegrationTests {
+
+	@Autowired
+	PollableFileSource<File> pollableFileSource;
+
+	private static File inputDir;
+
+	@BeforeClass
+	public static void setupInputDir() {
+		inputDir = new File(System.getProperty("java.io.tmpdir") + "/"
+				+ PollableFileSourceIntegrationTests.class.getSimpleName());
+		inputDir.mkdir();
+	}
+
+	@Before
+	public void generateTestFiles() throws Exception {
+		File.createTempFile("test", null, inputDir);
+		File.createTempFile("test", null, inputDir);
+		File.createTempFile("test", null, inputDir);
+	}
+
+	@After
+	public void resetTimestamp() {
+		((AtomicLong) new DirectFieldAccessor(pollableFileSource).getPropertyValue("lastListTimestamp")).set(System
+				.currentTimeMillis() - 5000);
+	}
+
+	@After
+	public void cleanoutInputDir() throws Exception {
+		File[] listFiles = inputDir.listFiles();
+		for (int i = 0; i < listFiles.length; i++) {
+			listFiles[i].delete();
+		}
+	}
+
+	@AfterClass
+	public static void removeInputDir() throws Exception {
+		inputDir.delete();
+	}
+
+	@Test
+	public void configured() throws Exception {
+		DirectFieldAccessor accessor = new DirectFieldAccessor(pollableFileSource);
+		assertEquals(inputDir, accessor.getPropertyValue("inputDirectory"));
+	}
+
+	@Test
+	public void getFiles() throws Exception {
+		Thread.sleep(1000);
+		Message<?> received1 = pollableFileSource.receive();
+		assertNotNull(received1);
+		pollableFileSource.onSend(received1);
+		Message<?> received2 = pollableFileSource.receive();
+		assertNotNull(received2);
+		pollableFileSource.onSend(received2);
+		Message<?> received3 = pollableFileSource.receive();
+		assertNotNull(received3);
+		pollableFileSource.onSend(received3);
+		assertNotSame(received1 + " == " + received2, received1.getPayload(), received2.getPayload());
+		assertNotSame(received1 + " == " + received3, received1.getPayload(), received3.getPayload());
+		assertNotSame(received2 + " == " + received3, received2.getPayload(), received3.getPayload());
+	}
+
+	@Test
+	public void parallelRetrieval() throws Exception {
+		Message<File> received1 = pollableFileSource.receive();
+		Message<File> received2 = pollableFileSource.receive();
+		Message<File> received3 = pollableFileSource.receive();
+		assertNotSame(received1 + " == " + received2, received1, received2);
+		assertNotSame(received1 + " == " + received3, received1, received3);
+		assertNotSame(received2 + " == " + received3, received2, received3);
+	}
+
+	@Test(timeout = 10000)
+	@Repeat(100)
+	public void concurrentProcessing() throws Exception {
+		CountDownLatch go = new CountDownLatch(1);
+		Runnable succesfulConsumer = new Runnable() {
+			public void run() {
+				Message<File> received = pollableFileSource.receive();
+				while (received == null) {
+					Thread.yield();
+					received = pollableFileSource.receive();
+				}
+				pollableFileSource.onSend(received);
+			}
+		};
+		Runnable failingConsumer = new Runnable() {
+			public void run() {
+				Message<File> received = pollableFileSource.receive();
+				if (received != null) {
+					pollableFileSource.onFailure(received, new RuntimeException("nothing"));
+				}
+			}
+		};
+		CountDownLatch succesfulDone = doConcurrently(3, succesfulConsumer, go);
+		CountDownLatch failingDone = doConcurrently(10, failingConsumer, go);
+		go.countDown();
+		succesfulDone.await();
+		failingDone.await();
+		// make sure three different files were taken
+		Message<File> received = pollableFileSource.receive();
+		if (received != null) {
+			pollableFileSource.onSend(received);
+		}
+		assertNull(received);
+	}
+
+	/**
+	 * Convenience method to run part of a test concurrently in multiple threads
+	 * 
+	 * @param numberOfThreads
+	 * @param todo the runnable that should be run by all the threads
+	 * @return a latch that will be counted down once all threads have run their
+	 * runnable.
+	 */
+	private CountDownLatch doConcurrently(int numberOfThreads, final Runnable todo, final CountDownLatch start) {
+		final CountDownLatch started = new CountDownLatch(numberOfThreads);
+		final CountDownLatch done = new CountDownLatch(numberOfThreads);
+		for (int i = 0; i < numberOfThreads; i++) {
+			new Thread(new Runnable() {
+
+				public void run() {
+					started.countDown();
+					try {
+						started.await();
+						start.await();
+					}
+					catch (InterruptedException e) {
+						Thread.currentThread().interrupt();
+					}
+					todo.run();
+					done.countDown();
+				}
+			}).start();
+		}
+		return done;
+	}
+}

org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceTests.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.adapter.file.replacement;
+
+import java.io.File;
+import java.io.FileFilter;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+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
+ */
+@SuppressWarnings("unchecked")
+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 };
+
+	@Before
+	public void initialize() throws Exception {
+		// inputDirectory = File.createTempFile("inputDir", null);
+		this.pollableFileSource = new PollableFileSource();
+		pollableFileSource.setInputDirectory(inputDirectory);
+		pollableFileSource.setMessageCreator(messageCreatorMock);
+		pollableFileSource.setInputDirectory(inputDirectoryMock);
+	}
+
+	@Before
+	public void setDefaultExpectations() {
+		expect(fileMock.exists()).andReturn(true).anyTimes();
+	}
+
+	@Test
+	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);
+		expect(fileMock.exists()).andReturn(true);
+		replay(allMocks);
+		pollableFileSource.onSend(pollableFileSource.receive());
+	}
+
+	@Test(expected = MessagingException.class)
+	public void requeueOnException() throws Exception {
+		expect(inputDirectoryMock.listFiles(isA(FileFilter.class))).andReturn(new File[] { fileMock });
+		expect(messageCreatorMock.createMessage(isA(File.class))).andThrow(new RuntimeException("just testing"));
+		replay(allMocks);
+		pollableFileSource.receive();
+		fail();
+	}
+
+	@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();
+		pollableFileSource.onFailure(received, new RuntimeException("failed"));
+		assertEquals(received, pollableFileSource.receive());
+	}
+
+	@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());
+		assertNull(pollableFileSource.receive());
+	}
+	
+	@After
+	public void verifyAll() {
+		verify(allMocks);
+	}
+
+}