diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/CompositeFileFilter.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/CompositeFileFilter.java deleted file mode 100644 index 60c672717a..0000000000 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/CompositeFileFilter.java +++ /dev/null @@ -1,59 +0,0 @@ -/* - * 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 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.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/ModificationTimeFileFilter.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/ModificationTimeFileFilter.java deleted file mode 100644 index 0c04335979..0000000000 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/ModificationTimeFileFilter.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * 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(); - } - -} diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/PollableFileSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/PollableFileSource.java deleted file mode 100644 index 28570a96ff..0000000000 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/replacement/PollableFileSource.java +++ /dev/null @@ -1,181 +0,0 @@ -/* - * 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 - */ -public class PollableFileSource implements PollableSource, MessageDeliveryAware, InitializingBean { - - private Log log = LogFactory.getLog(this.getClass()); - - private Queue queue = new PriorityBlockingQueue(); - - private MessageCreator messageCreator; - - private 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 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; - } - - /** - * 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 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("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); - } - } -} diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/CompositeFileFilterTest.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/CompositeFileFilterTest.java deleted file mode 100644 index df8ccd77be..0000000000 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/CompositeFileFilterTest.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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); - } -} diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/ModificationTimeFileFilterTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/ModificationTimeFileFilterTests.java deleted file mode 100644 index 20d9ee15f2..0000000000 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/ModificationTimeFileFilterTests.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * 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)); - } -} diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceIntegrationTests-context.xml b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceIntegrationTests-context.xml deleted file mode 100644 index 659b517216..0000000000 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceIntegrationTests-context.xml +++ /dev/null @@ -1,15 +0,0 @@ - - - - - - - - \ No newline at end of file diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceIntegrationTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceIntegrationTests.java deleted file mode 100644 index 8feba63593..0000000000 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceIntegrationTests.java +++ /dev/null @@ -1,183 +0,0 @@ -/* - * 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 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.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceTests.java deleted file mode 100644 index 1f2c9cb9b9..0000000000 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/replacement/PollableFileSourceTests.java +++ /dev/null @@ -1,112 +0,0 @@ -/* - * 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); - } - -} 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 index 5a632c1b16..51e0503571 100644 --- 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 @@ -13,29 +13,32 @@ * 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.Collection; +import java.util.HashSet; +import java.util.Set; /** - * Composition that delegates to multiple {@link FileFilter}s. The composition + * 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. + * for a file to be accepted by the composite. * * @author Iwein Fuld */ -public class CompositeFileFilter implements FileFilter { - - private final List fileFilters; +class CompositeFileFilter implements FileFilter { + private final Set fileFilters; public CompositeFileFilter(FileFilter... fileFilters) { - this.fileFilters = new ArrayList(Arrays.asList(fileFilters)); + this.fileFilters = new HashSet(Arrays.asList(fileFilters)); + } + + public CompositeFileFilter(HashSet fileFilters) { + this.fileFilters = new HashSet(fileFilters); } /** @@ -50,12 +53,25 @@ public class CompositeFileFilter implements FileFilter { return true; } - public void addFilter(FileFilter... filters) { - fileFilters.addAll(Arrays.asList(filters)); + /** + * @see #addFilters(Collection) + * @param filters one or more new filters to be used + * @return a new CompositeFileFilter with the additional filters + */ + public CompositeFileFilter addFilter(FileFilter... filters) { + return addFilters(Arrays.asList(filters)); } - public static CompositeFileFilter with(FileFilter... fileFilters) { - return new CompositeFileFilter(fileFilters); + /** + * Creates a new CompositeFileFilter that delegates to both the filters of + * this Composite and the new filters passed in. + * + * @param filtersToAdd + * @return a new CompositeFileFilter with the added filters + */ + public CompositeFileFilter addFilters(Collection filtersToAdd) { + HashSet newFilterSet = new HashSet(filtersToAdd); + newFilterSet.addAll(fileFilters); + return new CompositeFileFilter(newFilterSet); } - } 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 deleted file mode 100644 index 0f0e5c7b5d..0000000000 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/ModificationTimeFileFilter.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * 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 index 193f2655db..57344f916d 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 @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.integration.file; import java.io.File; @@ -29,7 +28,6 @@ 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; @@ -37,34 +35,41 @@ 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. + * 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. * * @author Iwein Fuld * - * @param + * @param the class of the payload of the message received from this + * {@link #PollableFileSource()} */ public class PollableFileSource implements PollableSource, MessageDeliveryAware, InitializingBean { - private final Log log = LogFactory.getLog(this.getClass()); + private static Log log = LogFactory.getLog(PollableFileSource.class); - private volatile Queue queue = new PriorityBlockingQueue(); + private volatile Queue fileQueue = new PriorityBlockingQueue(); private volatile MessageCreator messageCreator; private volatile File inputDirectory; - private final Map, File> tracker = new ConcurrentHashMap, File>(); + private final Map, File> messagesInFlight = new ConcurrentHashMap, File>(); - private final AtomicLong lastListTimestamp = new AtomicLong(); + private final AtomicLong currentListTimestamp = new AtomicLong(); - private final CompositeFileFilter filter = - CompositeFileFilter.with(new ModificationTimeFileFilter(lastListTimestamp)); + private final AtomicLong previousListTimestamp = new AtomicLong(); + private volatile CompositeFileFilter filter = new CompositeFileFilter(new ModificationTimeFileFilter()); + // Setters public void setQueue(Queue queue) { - this.queue = queue; + this.fileQueue = queue; } public void setMessageCreator(MessageCreator messageCreator) { @@ -75,109 +80,130 @@ public class PollableFileSource implements PollableSource, MessageDelivery this.inputDirectory = inputDirectory; } - public void setFilter(FileFilter filter) { - this.filter.addFilter(filter); + public void setFilter(FileFilter... filters) { + Assert.notEmpty(filters); + this.filter = filter.addFilter(filters); } public void afterPropertiesSet() throws Exception { if (this.messageCreator == null) { - throw new ConfigurationException(MessageCreator.class.getSimpleName() + " is required"); + throw new ConfigurationException(MessageCreator.class.getSimpleName() + "is required."); } if (this.inputDirectory == null) { - throw new ConfigurationException("inputDirectory is required"); + throw new ConfigurationException("inputDirectory cannot be null"); } if (!this.inputDirectory.exists()) { - throw new ConfigurationException("inputDirectory [" + inputDirectory + "] does not exist"); + throw new ConfigurationException(inputDirectory + " doesn't exist."); } if (!this.inputDirectory.canRead()) { - throw new ConfigurationException("unable to read from inputDirector [" + inputDirectory + "]"); + throw new ConfigurationException("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. + */ 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(); - } + traceState(); + refreshQueue(); + Message message = null; + File file = fileQueue.poll(); + // we cannot rely on isEmpty, so we have to do a null check if (file != null) { + message = createAndTrackMessage(file); if (log.isInfoEnabled()) { - log.info("preparing to send a message for file: [" + file + "]"); + log.info("Created message: [" + message + "]"); } - return createAndTrackMessage(file); } - // queue is empty - return null; + traceState(); + 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); - tracker.put(message, file); + messagesInFlight.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() + "]"); + fileQueue.add(file); + throw new MessagingException("Error creating message for file: [" + file + "]", e); } return message; } - private void refreshQueue(Queue queue) { + private void refreshQueue() { File[] freshFiles = getFreshFilesAndIncrementTimestamp(); - if (freshFiles != null) { + if (!ObjectUtils.isEmpty(freshFiles)) { 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); + freshFilesList.removeAll(fileQueue); + freshFilesList.removeAll(messagesInFlight.values()); + fileQueue.addAll(freshFilesList); if (log.isDebugEnabled()) { - log.debug("added to queue: " + freshFilesList); + log.debug("Added to queue: " + freshFilesList); } } } /* - * This is synchronized to prevent concurrent listings from causing - * duplication. + * 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. */ 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); + previousListTimestamp.set(currentListTimestamp.getAndSet(System.currentTimeMillis() / 1000 * 1000)); + File[] freshFiles = new File[] {}; + if (currentListTimestamp.get() > previousListTimestamp.get()) { + freshFiles = inputDirectory.listFiles(filter); } - tracker.remove(sentMessage); + 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} + * synchronized on this instance externally. */ public void onFailure(Message failedMessage, Throwable t) { - log.warn("not sent: " + failedMessage); - queue.add(tracker.get(failedMessage)); - tracker.remove(failedMessage); + log.warn("Failed to send: " + failedMessage); + fileQueue.add(messagesInFlight.get(failedMessage)); + messagesInFlight.remove(failedMessage); } + public void onSend(Message sentMessage) { + if (log.isDebugEnabled()) { + log.debug("Sent: " + sentMessage); + } + messagesInFlight.remove(sentMessage); + } + + /* + * utility method to trace the stateful collections of this instance. + */ + private void traceState() { + if (log.isTraceEnabled()) { + log.trace("Files to be received: [" + fileQueue + "]"); + log.trace("Messages in flight: [" + messagesInFlight.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/CompositeFileFilterTest.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/CompositeFileFilterTest.java index f9e5645ae8..c822eeef83 100644 --- 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 @@ -13,19 +13,12 @@ * 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 static org.easymock.classextension.EasyMock.*; +import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; @@ -35,44 +28,40 @@ import org.junit.Test; */ public class CompositeFileFilterTest { - private FileFilter fileFilterMock = createMock(FileFilter.class); - - private CompositeFileFilter compositeFileFilter; + private FileFilter fileFilterMock1 = createMock(FileFilter.class); + private FileFilter fileFilterMock2 = createMock(FileFilter.class); 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); + CompositeFileFilter compositeFileFilter = new CompositeFileFilter(fileFilterMock1, fileFilterMock2); + expect(fileFilterMock1.accept(fileMock)).andReturn(true).times(1); + expect(fileFilterMock2.accept(fileMock)).andReturn(true).times(1); + replay(fileFilterMock1, fileFilterMock2); assertTrue(compositeFileFilter.accept(fileMock)); - verify(fileFilterMock); + verify(fileFilterMock1, fileFilterMock2); } @Test public void forwardedToAddedFilters() throws Exception { - expect(fileFilterMock.accept(fileMock)).andReturn(true).times(4); - replay(fileFilterMock); - compositeFileFilter.addFilter(fileFilterMock, fileFilterMock); + CompositeFileFilter compositeFileFilter = new CompositeFileFilter().addFilter(fileFilterMock1, fileFilterMock2); + expect(fileFilterMock1.accept(fileMock)).andReturn(true).times(1); + expect(fileFilterMock2.accept(fileMock)).andReturn(true).times(1); + replay(fileFilterMock1, fileFilterMock2); assertTrue(compositeFileFilter.accept(fileMock)); - verify(fileFilterMock); + verify(fileFilterMock1, fileFilterMock2); } @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); + public void negative() throws Exception { + CompositeFileFilter compositeFileFilter = new CompositeFileFilter(fileFilterMock1, fileFilterMock2); + expect(fileFilterMock1.accept(fileMock)).andReturn(false).times(1); + expect(fileFilterMock2.accept(fileMock)).andReturn(true).anyTimes(); + replay(fileFilterMock1, fileFilterMock2); assertFalse(compositeFileFilter.accept(fileMock)); - verify(fileFilterMock); + verify(fileFilterMock1, fileFilterMock2); } - } 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 deleted file mode 100644 index 576f03b0d0..0000000000 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/ModificationTimeFileFilterTests.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * 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 index 5380c62bf1..1be0261344 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 @@ -1,17 +1,17 @@ + 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"> - - - - - + class="org.springframework.integration.file.PollableFileSource" + p:inputDirectory="${java.io.tmpdir}/PollableFileSourceIntegrationTests" + p:messageCreator-ref="messageCreator" p:filter-ref="someFilter" /> + + + + \ 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 94396f0fb8..f99e0b17fc 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 @@ -13,9 +13,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - 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; @@ -31,7 +31,6 @@ 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; @@ -51,7 +50,6 @@ public class PollableFileSourceIntegrationTests { private static File inputDir; - @BeforeClass public static void setupInputDir() { inputDir = new File(System.getProperty("java.io.tmpdir") + "/" @@ -61,16 +59,15 @@ public class PollableFileSourceIntegrationTests { @Before public void generateTestFiles() throws Exception { - File.createTempFile("test", null, inputDir); - File.createTempFile("test", null, inputDir); - File.createTempFile("test", null, inputDir); + File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); + File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); + File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); } @After - public void resetTimestamp() { - ((AtomicLong) new DirectFieldAccessor(pollableFileSource) - .getPropertyValue("lastListTimestamp")) - .set(System.currentTimeMillis() - 5000); + public void resetTimestamps() { + ((AtomicLong) new DirectFieldAccessor(pollableFileSource).getPropertyValue("previousListTimestamp")).set(0); + ((AtomicLong) new DirectFieldAccessor(pollableFileSource).getPropertyValue("currentListTimestamp")).set(0); } @After @@ -86,7 +83,6 @@ public class PollableFileSourceIntegrationTests { inputDir.delete(); } - @Test public void configured() throws Exception { DirectFieldAccessor accessor = new DirectFieldAccessor(pollableFileSource); @@ -95,9 +91,8 @@ public class PollableFileSourceIntegrationTests { @Test public void getFiles() throws Exception { - Thread.sleep(1000); Message received1 = pollableFileSource.receive(); - assertNotNull(received1); + assertNotNull("This should return the first message", received1); pollableFileSource.onSend(received1); Message received2 = pollableFileSource.receive(); assertNotNull(received2); @@ -120,8 +115,20 @@ public class PollableFileSourceIntegrationTests { assertNotSame(received2 + " == " + received3, received2, received3); } - @Test(timeout = 10000) - @Repeat(100) + @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 { CountDownLatch go = new CountDownLatch(1); Runnable succesfulConsumer = new Runnable() { @@ -145,8 +152,13 @@ public class PollableFileSourceIntegrationTests { CountDownLatch succesfulDone = doConcurrently(3, succesfulConsumer, go); CountDownLatch failingDone = doConcurrently(10, failingConsumer, go); go.countDown(); - succesfulDone.await(); - failingDone.await(); + try { + succesfulDone.await(); + failingDone.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } // make sure three different files were taken Message received = pollableFileSource.receive(); if (received != null) { @@ -160,13 +172,15 @@ public class PollableFileSourceIntegrationTests { * * @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. + * @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 { @@ -183,5 +197,4 @@ public class PollableFileSourceIntegrationTests { } 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 index b013841f17..378187b7a2 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 @@ -13,30 +13,22 @@ * 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 java.util.concurrent.atomic.AtomicLong; -import org.junit.After; 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 */ @@ -57,8 +49,8 @@ public class PollableFileSourceTests { private File fileMock = createMock(File.class); - private Object[] allMocks = new Object[] { inputDirectoryMock, messageCreatorMock, messageMock, filterMock, fileMock }; - + private Object[] allMocks = new Object[] { inputDirectoryMock, messageCreatorMock, messageMock, filterMock, + fileMock }; @Before public void initialize() throws Exception { @@ -74,24 +66,33 @@ public class PollableFileSourceTests { 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()); + verify(allMocks); } - @Test(expected = MessagingException.class) + @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); - pollableFileSource.receive(); - fail(); + try { + pollableFileSource.receive(); + fail(); + } + catch (MessagingException e) { + // ok + } + resetTimestamp(pollableFileSource); + assertSame(messageMock, pollableFileSource.receive()); + verify(allMocks); } @Test @@ -101,8 +102,14 @@ public class PollableFileSourceTests { 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()); + verify(allMocks); + } + + private void resetTimestamp(PollableFileSource pollableFileSource) throws Exception { + ((AtomicLong) new DirectFieldAccessor(pollableFileSource).getPropertyValue("currentListTimestamp")).set(0); } @Test @@ -112,12 +119,8 @@ public class PollableFileSourceTests { expect(messageCreatorMock.createMessage(fileMock)).andReturn(messageMock); replay(allMocks); assertEquals(messageMock, pollableFileSource.receive()); + resetTimestamp(pollableFileSource); assertNull(pollableFileSource.receive()); - } - - @After - public void verifyAll() { verify(allMocks); } - } diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/TestFileFilter.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/TestFileFilter.java new file mode 100644 index 0000000000..8e5e096e6e --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/TestFileFilter.java @@ -0,0 +1,12 @@ +package org.springframework.integration.file; + +import java.io.File; +import java.io.FileFilter; + +public class TestFileFilter implements FileFilter{ + + public boolean accept(File pathname) { + return true; + } + +}