diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/CompositeFileFilter.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/CompositeFileFilter.java new file mode 100644 index 0000000000..5a632c1b16 --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/CompositeFileFilter.java @@ -0,0 +1,61 @@ +/* + * 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.file; + +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 composition + * 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 fileFilters; + + + public CompositeFileFilter(FileFilter... fileFilters) { + this.fileFilters = new ArrayList(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); + } + +} 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..0f0e5c7b5d --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/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.file; + +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(); + } + +} 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 new file mode 100644 index 0000000000..193f2655db --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.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.file; + +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 + */ +public class PollableFileSource implements PollableSource, MessageDeliveryAware, InitializingBean { + + private final Log log = LogFactory.getLog(this.getClass()); + + private volatile Queue queue = new PriorityBlockingQueue(); + + private volatile MessageCreator messageCreator; + + private volatile File inputDirectory; + + private final Map, File> tracker = new ConcurrentHashMap, File>(); + + private final AtomicLong lastListTimestamp = new AtomicLong(); + + private final CompositeFileFilter filter = + CompositeFileFilter.with(new ModificationTimeFileFilter(lastListTimestamp)); + + + public void setQueue(Queue queue) { + this.queue = queue; + } + + public void setMessageCreator(MessageCreator 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(MessageCreator.class.getSimpleName() + " is required"); + } + if (this.inputDirectory == null) { + throw new ConfigurationException("inputDirectory is required"); + } + if (!this.inputDirectory.exists()) { + throw new ConfigurationException("inputDirectory [" + inputDirectory + "] does not exist"); + } + if (!this.inputDirectory.canRead()) { + throw new ConfigurationException("unable to read from inputDirector [" + inputDirectory + "]"); + } + } + + public Message 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 createAndTrackMessage(File file) throws MessagingException { + Message 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 queue) { + File[] freshFiles = getFreshFilesAndIncrementTimestamp(); + if (freshFiles != null) { + List freshFilesList = new ArrayList(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; + } + + public void onSend(Message sentMessage) { + if (log.isDebugEnabled()) { + log.debug("sent: " + sentMessage); + } + tracker.remove(sentMessage); + } + + /** + * 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); + } + +} diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/CompositeFileFilterTest.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/CompositeFileFilterTest.java new file mode 100644 index 0000000000..f9e5645ae8 --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/CompositeFileFilterTest.java @@ -0,0 +1,78 @@ +/* + * 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.file; + +import static org.easymock.EasyMock.expect; +import static org.easymock.classextension.EasyMock.createMock; +import static org.easymock.classextension.EasyMock.createNiceMock; +import static org.easymock.classextension.EasyMock.replay; +import static org.easymock.classextension.EasyMock.verify; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.FileFilter; + +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); + } + +} diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/ModificationTimeFileFilterTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/ModificationTimeFileFilterTests.java new file mode 100644 index 0000000000..576f03b0d0 --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/ModificationTimeFileFilterTests.java @@ -0,0 +1,50 @@ +/* + * 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.file; + +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)); + } + +} 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 new file mode 100644 index 0000000000..5380c62bf1 --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests-context.xml @@ -0,0 +1,17 @@ + + + + + + + + + + \ 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 new file mode 100644 index 0000000000..94396f0fb8 --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceIntegrationTests.java @@ -0,0 +1,187 @@ +/* + * 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.file; + +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 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 received1 = pollableFileSource.receive(); + Message received2 = pollableFileSource.receive(); + Message 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 received = pollableFileSource.receive(); + while (received == null) { + Thread.yield(); + received = pollableFileSource.receive(); + } + pollableFileSource.onSend(received); + } + }; + Runnable failingConsumer = new Runnable() { + public void run() { + Message 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 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; + } + +} 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 new file mode 100644 index 0000000000..b013841f17 --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/PollableFileSourceTests.java @@ -0,0 +1,123 @@ +/* + * 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.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 static org.junit.Assert.fail; + +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; + +/** + * @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); + } + +}