diff --git a/org.springframework.integration.file/.classpath b/org.springframework.integration.file/.classpath index 0ed5a603da..95d6d53dcd 100644 --- a/org.springframework.integration.file/.classpath +++ b/org.springframework.integration.file/.classpath @@ -19,6 +19,7 @@ + diff --git a/org.springframework.integration.file/ivy.xml b/org.springframework.integration.file/ivy.xml index 0b9c4132e4..ffad6a5019 100644 --- a/org.springframework.integration.file/ivy.xml +++ b/org.springframework.integration.file/ivy.xml @@ -28,6 +28,7 @@ + diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 7ff4fc6d8e..99342edcbb 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -16,27 +16,23 @@ package org.springframework.integration.file; -import java.io.File; -import java.io.IOException; -import java.util.Comparator; -import java.util.HashSet; -import java.util.List; -import java.util.Queue; -import java.util.Set; -import java.util.concurrent.PriorityBlockingQueue; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; import org.springframework.integration.aggregator.Resequencer; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessagingException; +import org.springframework.integration.file.locking.FileLocker; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageSource; import org.springframework.util.Assert; +import java.io.File; +import java.io.IOException; +import java.util.*; +import java.util.concurrent.PriorityBlockingQueue; + /** * {@link MessageSource} that creates messages from a file system directory. To * prevent messages for certain files, you may supply a {@link FileListFilter}. @@ -59,177 +55,208 @@ import org.springframework.util.Assert; *

* FileReadingMessageSource is fully thread-safe under concurrent * receive() invocations and message delivery callbacks. - * + * * @author Iwein Fuld * @author Mark Fisher */ public class FileReadingMessageSource implements MessageSource, - InitializingBean { + InitializingBean { - private static final int INTERNAL_QUEUE_CAPACITY = 5; + private static final int INTERNAL_QUEUE_CAPACITY = 5; - private static final Log logger = LogFactory - .getLog(FileReadingMessageSource.class); + private static final Log logger = LogFactory + .getLog(FileReadingMessageSource.class); - private volatile File inputDirectory; + private volatile File inputDirectory; - private volatile boolean autoCreateDirectory = true; + private volatile boolean autoCreateDirectory = true; - /** - * {@link PriorityBlockingQueue#iterator()} throws - * {@link java.util.ConcurrentModificationException} in Java 5. There is no - * locking around the queue, so there is also no iteration. - */ - private final Queue toBeReceived; + /** + * {@link PriorityBlockingQueue#iterator()} throws + * {@link java.util.ConcurrentModificationException} in Java 5. There is no + * locking around the queue, so there is also no iteration. + */ + private final Queue toBeReceived; - private volatile FileListFilter filter = new AcceptOnceFileListFilter(); + private volatile FileListFilter filter = new AcceptOnceFileListFilter(); - private boolean scanEachPoll = false; + private volatile FileLocker locker = new NoopFileLocker(); - /** - * Creates a FileReadingMessageSource with a naturally ordered queue. - */ - public FileReadingMessageSource() { - toBeReceived = new PriorityBlockingQueue(INTERNAL_QUEUE_CAPACITY); - } + private boolean scanEachPoll = false; - /** - * Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue} - * ordered with the passed in {@link Comparator} - * - * No guarantees about file delivery order can be made under concurrent - * access. - */ - public FileReadingMessageSource(Comparator receptionOrderComparator) { - toBeReceived = new PriorityBlockingQueue(INTERNAL_QUEUE_CAPACITY, - receptionOrderComparator); - } + /** + * Creates a FileReadingMessageSource with a naturally ordered queue. + */ + public FileReadingMessageSource() { + toBeReceived = new PriorityBlockingQueue(INTERNAL_QUEUE_CAPACITY); + } - /** - * Specify the input directory. - */ - public void setInputDirectory(Resource inputDirectory) { - Assert.notNull(inputDirectory, "inputDirectory must not be null"); - try { - this.inputDirectory = inputDirectory.getFile(); - } catch (IOException ioe) { - try { - // fallback to the URI - this.inputDirectory = new File(inputDirectory.getURI()); - } catch (Exception e) { - throw new IllegalArgumentException( - "Unexpected IOException when looking for source directory: " - + inputDirectory, ioe); - } - } - } + /** + * Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue} + * ordered with the passed in {@link Comparator} + *

+ * No guarantees about file delivery order can be made under concurrent + * access. + */ + public FileReadingMessageSource(Comparator receptionOrderComparator) { + toBeReceived = new PriorityBlockingQueue(INTERNAL_QUEUE_CAPACITY, + receptionOrderComparator); + } - /** - * Specify whether to create the source directory automatically if it does - * not yet exist upon initialization. By default, this value is - * true. If set to false and the - * source directory does not exist, an Exception will be thrown upon - * initialization. - */ - public void setAutoCreateDirectory(boolean autoCreateDirectory) { - this.autoCreateDirectory = autoCreateDirectory; - } + /** + * Specify the input directory. + */ + public void setInputDirectory(Resource inputDirectory) { + Assert.notNull(inputDirectory, "inputDirectory must not be null"); + try { + this.inputDirectory = inputDirectory.getFile(); + } catch (IOException ioe) { + try { + // fallback to the URI + this.inputDirectory = new File(inputDirectory.getURI()); + } catch (Exception e) { + throw new IllegalArgumentException( + "Unexpected IOException when looking for source directory: " + + inputDirectory, ioe); + } + } + } - /** - * Sets a {@link FileListFilter}. By default a - * {@link AcceptOnceFileListFilter} with no bounds is used. In most cases a - * customized {@link FileListFilter} will be needed to deal with - * modification and duplication concerns. If multiple filters are required a - * {@link CompositeFileListFilter} can be used to group them together. - *

- * The supplied filter must be thread safe.. - */ - public void setFilter(FileListFilter filter) { - Assert.notNull(filter, "'filter' should not be null"); - this.filter = filter; - } + /** + * Specify whether to create the source directory automatically if it does + * not yet exist upon initialization. By default, this value is + * true. If set to false and the + * source directory does not exist, an Exception will be thrown upon + * initialization. + */ + public void setAutoCreateDirectory(boolean autoCreateDirectory) { + this.autoCreateDirectory = autoCreateDirectory; + } - /** - * Optional. Set this flag if you want to make sure the internal queue is - * refreshed with the latest content of the input directory on each poll. - *

- * By default this implementation will empty its queue before looking at the - * directory again. In cases where order is relevant it is important to - * consider the effects of setting this flag. The internal - * {@link PriorityBlockingQueue} that this class is keeping will more likely - * be out of sync with the filesystem if this flag is set to - * false, but it will change more often (causing reordering) if - * it is set to true. - */ - public void setScanEachPoll(boolean scanEachPoll) { - this.scanEachPoll = scanEachPoll; - } + /** + * Sets a {@link FileListFilter}. By default a + * {@link AcceptOnceFileListFilter} with no bounds is used. In most cases a + * customized {@link FileListFilter} will be needed to deal with + * modification and duplication concerns. If multiple filters are required a + * {@link CompositeFileListFilter} can be used to group them together. + *

+ * The supplied filter must be thread safe.. + */ + public void setFilter(FileListFilter filter) { + Assert.notNull(filter, "'filter' must not be null"); + this.filter = filter; + } - public final void afterPropertiesSet() { - if (!this.inputDirectory.exists() && this.autoCreateDirectory) { - this.inputDirectory.mkdirs(); - } - Assert.isTrue(this.inputDirectory.exists(), "Source directory [" - + inputDirectory + "] does not exist."); - Assert.isTrue(this.inputDirectory.isDirectory(), "Source path [" - + this.inputDirectory + "] does not point to a directory."); - Assert.isTrue(this.inputDirectory.canRead(), "Source directory [" - + this.inputDirectory + "] is not readable."); - } + /** + * Optional. Sets a {@link org.springframework.integration.file.locking.FileLocker} + * to be used instead of the default NoopFileLocker. Note that the locker is not queried + * by this FileReadingMessageSource: integration with a FileListFilter is an external concern. + *

+ * The supplied FileLocker must be thread safe + */ + public void setLocker(FileLocker locker) { + Assert.notNull(locker, "'fileLocker' must not be null."); + this.locker = locker; + } - public Message receive() throws MessagingException { - Message message = null; - // rescan only if needed or explicitly configured - if (scanEachPoll || toBeReceived.isEmpty()) { - scanInputDirectory(); - } - File file = toBeReceived.poll(); - // we can't rely on isEmpty for concurrency reasons - if (file != null) { - message = MessageBuilder.withPayload(file).build(); - if (logger.isInfoEnabled()) { - logger.info("Created message: [" + message + "]"); - } - } - return message; - } + /** + * Optional. Set this flag if you want to make sure the internal queue is + * refreshed with the latest content of the input directory on each poll. + *

+ * By default this implementation will empty its queue before looking at the + * directory again. In cases where order is relevant it is important to + * consider the effects of setting this flag. The internal + * {@link PriorityBlockingQueue} that this class is keeping will more likely + * be out of sync with the filesystem if this flag is set to + * false, but it will change more often (causing reordering) if + * it is set to true. + */ + public void setScanEachPoll(boolean scanEachPoll) { + this.scanEachPoll = scanEachPoll; + } - private void scanInputDirectory() { - File[] fileArray = inputDirectory.listFiles(); - if (fileArray == null) { - throw new MessagingException( - "Either the path [" - + this.inputDirectory - + "] does not denote a directory, or an I/O error has occured."); - } - List filteredFiles = this.filter.filterFiles(fileArray); - Set freshFiles = new HashSet(filteredFiles); - if (!freshFiles.isEmpty()) { - toBeReceived.addAll(freshFiles); - if (logger.isDebugEnabled()) { - logger.debug("Added to queue: " + freshFiles); - } - } - } + public final void afterPropertiesSet() { + if (!this.inputDirectory.exists() && this.autoCreateDirectory) { + this.inputDirectory.mkdirs(); + } + Assert.isTrue(this.inputDirectory.exists(), "Source directory [" + + inputDirectory + "] does not exist."); + Assert.isTrue(this.inputDirectory.isDirectory(), "Source path [" + + this.inputDirectory + "] does not point to a directory."); + Assert.isTrue(this.inputDirectory.canRead(), "Source directory [" + + this.inputDirectory + "] is not readable."); + } - /** - * Adds the failed message back to the 'toBeReceived' queue. - */ - public void onFailure(Message failedMessage, Throwable t) { - if (logger.isWarnEnabled()) { - logger.warn("Failed to send: " + failedMessage); - } - toBeReceived.offer(failedMessage.getPayload()); - } + public Message receive() throws MessagingException { + Message message = null; + // rescan only if needed or explicitly configured + if (scanEachPoll || toBeReceived.isEmpty()) { + scanInputDirectory(); + } + File file = toBeReceived.poll(); + // file == null means the queue was empty + // we can't rely on isEmpty for concurrency reasons + while (file != null && !locker.lock(file)) { + file = toBeReceived.poll(); + } + if (file != null) { + message = MessageBuilder.withPayload(file).build(); + if (logger.isInfoEnabled()) { + logger.info("Created message: [" + message + "]"); + } + } + return message; + } - /** - * The message is just logged. It was already removed from the queue during - * the call to receive() - */ - public void onSend(Message sentMessage) { - if (logger.isDebugEnabled()) { - logger.debug("Sent: " + sentMessage); - } - } + private void scanInputDirectory() { + File[] fileArray = inputDirectory.listFiles(); + if (fileArray == null) { + throw new MessagingException( + "The path [" + + this.inputDirectory + + "] does not denote a properly accessible directory."); + } + List filteredFiles = this.filter.filterFiles(fileArray); + Set freshFiles = new HashSet(filteredFiles); + if (!freshFiles.isEmpty()) { + toBeReceived.addAll(freshFiles); + if (logger.isDebugEnabled()) { + logger.debug("Added to queue: " + freshFiles); + } + } + } + /** + * Adds the failed message back to the 'toBeReceived' queue. + */ + public void onFailure(Message failedMessage, Throwable t) { + if (logger.isWarnEnabled()) { + logger.warn("Failed to send: " + failedMessage); + } + toBeReceived.offer(failedMessage.getPayload()); + } + + /** + * The message is just logged. It was already removed from the queue during + * the call to receive() + */ + public void onSend(Message sentMessage) { + if (logger.isDebugEnabled()) { + logger.debug("Sent: " + sentMessage); + } + } + + /** + * Implementation of FileLocker that doesn't provide any protection against duplicate listing. + */ + class NoopFileLocker implements FileLocker { + + public boolean lock(File fileToLock) { + return true; + } + + public void unlock(File fileToUnlock) { + //noop + } + } } diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/NoopFileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/NoopFileLocker.java new file mode 100644 index 0000000000..d8fc7fedbb --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/NoopFileLocker.java @@ -0,0 +1,38 @@ +/* + * 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 org.springframework.integration.file.locking.FileLocker; + +import java.io.File; + +/** + * Implementation of FileLocker that doesn't provide any protection against duplicate listing. This is the default used + * by the FileReadingMessageSource. + * + * @author Iwein Fuld + */ +final class NoopFileLocker implements FileLocker { + + public boolean lock(File fileToLock) { + return true; + } + + public void unlock(File fileToUnlock) { + //noop + } +} + diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java new file mode 100644 index 0000000000..30b8ec1bae --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java @@ -0,0 +1,38 @@ +/* + * 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.locking; + +import java.io.File; + +/** + * A FileLocker is a strategy that can ensure that files are only processed a single time. + * + * Implementations are free implement any relation between locking and unlocking (e.g. the Noop + * + * @author Iwein Fuld + */ +public interface FileLocker { + + /** + * Tries to lock the given file and returns true if it was successful, false otherwise. + */ + boolean lock(File fileToLock); + + /** + * Unlocks the given file. + */ + void unlock(File fileToUnlock); +} diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileListFilter.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileListFilter.java new file mode 100644 index 0000000000..5b0825c322 --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileListFilter.java @@ -0,0 +1,52 @@ +/* + * 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.locking; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.file.AbstractFileListFilter; +import static org.springframework.integration.file.locking.LockFileFileLocker.*; + +import java.io.File; + +/** + * LockFileFileListFilter keeps track of the files it should pick up by adding lock files in a directory. If this + * directory is the same as the directory of the original files addition precautions need to be taken to avoid treating + * the lock files as normal input. + * + * If different LockFileFileListFilters share their lock directory they are guaranteed not to pass the same file. + * + * @author Iwein Fuld + */ +public class LockFileFileListFilter extends AbstractFileListFilter { + private static final Log logger = LogFactory.getLog(LockFileFileListFilter.class); + + private File workdir; + + public LockFileFileListFilter(File workdir) { + this.workdir = workdir; + } + + protected boolean accept(File file) { + File lockFile = new File(workdir, file.getName()+ LOCK_SUFFIX); + File preLockFile = new File(workdir, file.getName()+ PRELOCK_SUFFIX); + if (lockFile.exists()||preLockFile.exists()){ + return false; + } + String name = file.getName(); + return !name.endsWith(LOCK_SUFFIX)&&!name.endsWith(PRELOCK_SUFFIX); + } +} diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileLocker.java new file mode 100644 index 0000000000..a9f6179a12 --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileLocker.java @@ -0,0 +1,79 @@ +/* + * 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.locking; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.util.Assert; + +import java.io.File; +import java.io.IOException; + +/** + * @author Iwein Fuld + */ +public class LockFileFileLocker implements FileLocker { + private static final Log logger = LogFactory.getLog(LockFileFileLocker.class); + + private File workdir; + static final String LOCK_SUFFIX = ".lock"; + static final String PRELOCK_SUFFIX = ".prelock"; + + public LockFileFileLocker(File workdir) { + Assert.notNull(workdir, "Work directy must not be null."); + Assert.isTrue(workdir.isDirectory(),"Work directy must be a directory."); + Assert.isTrue(workdir.canWrite(), "Work directy must be write accessible."); + this.workdir = workdir; + } + + /** + * Makes a best effort attempt at locking the file atomically. The chances of success are wholly dependant on the + * underlying operating system. + * + * The locking mechanism will create a prelock file, remove write permissions from that file and move it to a lock + * file location. + */ + public boolean lock(File fileToLock) { + File lockFile = new File(workdir, fileToLock.getName() + LOCK_SUFFIX); + File preLockFile = new File(workdir, fileToLock.getName() + PRELOCK_SUFFIX); + if (lockFile.exists() || preLockFile.exists()) { + return false; + } + try { + preLockFile.createNewFile(); + } catch (IOException e) { + logger.warn("Failed to lock file", e); + return false; + } + preLockFile.setWritable(false); + if (preLockFile.renameTo(lockFile)) { + return true; + } + preLockFile.delete(); + logger.warn("Failed to lock file [" + fileToLock + "]"); + return false; + } + + /** + * Sets the lockFile for this file to be writable and then deletes it. If a lock happens concurrently, the delete + * should not succeed, so the lock will remain in place. This might be different on various operating systems. + */ + public void unlock(File fileToUnlock) { + File lockFile = new File(workdir, fileToUnlock.getName() + LOCK_SUFFIX); + lockFile.setWritable(true); + lockFile.delete(); + } +} diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java index abadaa9998..7c41d1785e 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java @@ -16,22 +16,19 @@ package org.springframework.integration.file; -import static org.mockito.Mockito.*; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertSame; - -import java.io.File; -import java.util.Comparator; - +import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; +import static org.mockito.Mockito.*; import org.mockito.runners.MockitoJUnit44Runner; import org.springframework.core.io.Resource; import org.springframework.integration.core.Message; +import org.springframework.integration.file.locking.FileLocker; + +import java.io.File; +import java.util.Comparator; /** * @author Iwein Fuld @@ -41,94 +38,119 @@ import org.springframework.integration.core.Message; @RunWith(MockitoJUnit44Runner.class) public class FileReadingMessageSourceTests { - private FileReadingMessageSource source; + private FileReadingMessageSource source; - @Mock - private File inputDirectoryMock; + @Mock + private File inputDirectoryMock; - @Mock - private Resource inputDirectoryResourceMock; + @Mock + private Resource inputDirectoryResourceMock; - @Mock - private File fileMock; + @Mock + private File fileMock; - @Mock - private Comparator comparator; + @Mock + private FileLocker locker; - public void prepResource() throws Exception { - when(inputDirectoryResourceMock.exists()).thenReturn(true); - when(inputDirectoryResourceMock.getFile()).thenReturn(inputDirectoryMock); - when(inputDirectoryMock.canRead()).thenReturn(true); - } + @Mock + private Comparator comparator; - @Before - public void initialize() throws Exception { - prepResource(); - this.source = new FileReadingMessageSource(comparator); - source.setInputDirectory(inputDirectoryResourceMock); - } + public void prepResource() throws Exception { + when(inputDirectoryResourceMock.exists()).thenReturn(true); + when(inputDirectoryResourceMock.getFile()).thenReturn(inputDirectoryMock); + when(inputDirectoryMock.canRead()).thenReturn(true); + when(locker.lock(isA(File.class))).thenReturn(true); + } - @Test - public void straightProcess() throws Exception { - when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock }); - source.onSend(source.receive()); - } + @Before + public void initialize() throws Exception { + prepResource(); + this.source = new FileReadingMessageSource(comparator); + source.setInputDirectory(inputDirectoryResourceMock); + source.setLocker(locker); + } - @Test - public void requeueOnFailure() throws Exception { - when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock }); - Message received = source.receive(); - assertNotNull(received); - source.onFailure(received, new RuntimeException("failed")); - assertEquals(received.getPayload(), source.receive().getPayload()); - verify(inputDirectoryMock,times(1)).listFiles(); - } + @Test + public void straightProcess() throws Exception { + when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + source.onSend(source.receive()); + } - @Test - public void scanEachPoll() throws Exception { - File anotherFileMock = mock(File.class); - when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock, anotherFileMock }); - source.setScanEachPoll(true); - assertNotNull(source.receive()); - assertNotNull(source.receive()); - assertNull(source.receive()); - verify(inputDirectoryMock,times(3)).listFiles(); - } + @Test + public void requeueOnFailure() throws Exception { + when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + Message received = source.receive(); + assertNotNull(received); + source.onFailure(received, new RuntimeException("failed")); + assertEquals(received.getPayload(), source.receive().getPayload()); + verify(inputDirectoryMock, times(1)).listFiles(); + } - @Test - public void noDuplication() throws Exception { - when(inputDirectoryMock.listFiles()).thenReturn(new File[] { fileMock }); - Message received = source.receive(); - assertNotNull(received); - assertEquals(fileMock, received.getPayload()); - assertNull(source.receive()); - verify(inputDirectoryMock,times(2)).listFiles(); - } + @Test + public void scanEachPoll() throws Exception { + File anotherFileMock = mock(File.class); + when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock, anotherFileMock}); + source.setScanEachPoll(true); + assertNotNull(source.receive()); + assertNotNull(source.receive()); + assertNull(source.receive()); + verify(inputDirectoryMock, times(3)).listFiles(); + } - @Test(expected = IllegalArgumentException.class) - public void nullFilter() throws Exception { - source.setFilter(null); - } + @Test + public void noDuplication() throws Exception { + when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + Message received = source.receive(); + assertNotNull(received); + assertEquals(fileMock, received.getPayload()); + assertNull(source.receive()); + verify(inputDirectoryMock, times(2)).listFiles(); + } - @Test - public void orderedReception() throws Exception { - File file1 = mock(File.class); - File file2 = mock(File.class); - File file3 = mock(File.class); + @Test(expected = IllegalArgumentException.class) + public void nullFilter() throws Exception { + source.setFilter(null); + } - // record the comparator to reverse order the files - when(comparator.compare(file1, file2)).thenReturn(1); - when(comparator.compare(file1, file3)).thenReturn(1); - when(comparator.compare(file2, file3)).thenReturn(1); - when(comparator.compare(file2, file1)).thenReturn(-1); - when(comparator.compare(file3, file1)).thenReturn(-1); - when(comparator.compare(file3, file2)).thenReturn(-1); + @Test + public void lockIsAquired() { + when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + Message received = source.receive(); + assertNotNull(received); + assertEquals(fileMock, received.getPayload()); + verify(locker).lock(fileMock); + } - when(inputDirectoryMock.listFiles()).thenReturn(new File[] { file2, file3, file1 }); - assertSame(file3, source.receive().getPayload()); - assertSame(file2, source.receive().getPayload()); - assertSame(file1, source.receive().getPayload()); - assertNull(source.receive()); - verify(inputDirectoryMock,times(2)).listFiles(); - } + @Test + public void lockedFilesAreIgnored() { + when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); + when(locker.lock(fileMock)).thenReturn(false); + Message received = source.receive(); + assertNull(received); + verify(locker).lock(fileMock); + } + + + + @Test + public void orderedReception() throws Exception { + File file1 = mock(File.class); + File file2 = mock(File.class); + File file3 = mock(File.class); + + // record the comparator to reverse order the files + when(comparator.compare(file1, file2)).thenReturn(1); + when(comparator.compare(file1, file3)).thenReturn(1); + when(comparator.compare(file2, file3)).thenReturn(1); + when(comparator.compare(file2, file1)).thenReturn(-1); + when(comparator.compare(file3, file1)).thenReturn(-1); + when(comparator.compare(file3, file2)).thenReturn(-1); + + when(inputDirectoryMock.listFiles()).thenReturn(new File[]{file2, file3, file1}); + assertSame(file3, source.receive().getPayload()); + assertSame(file2, source.receive().getPayload()); + assertSame(file1, source.receive().getPayload()); + assertNull(source.receive()); + verify(inputDirectoryMock, times(2)).listFiles(); + } } diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml new file mode 100644 index 0000000000..b7a1ba5463 --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests.java new file mode 100644 index 0000000000..eb966070ce --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests.java @@ -0,0 +1,76 @@ +/* + * 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.locking; + +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.file.FileReadingMessageSource; +import org.springframework.integration.test.matcher.PayloadMatcher; +import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload; +import org.junit.runner.RunWith; +import org.junit.Test; +import org.junit.BeforeClass; +import org.junit.Before; +import org.junit.Assert; +import org.junit.matchers.JUnitMatchers; +import static org.junit.Assert.assertThat; +import org.hamcrest.CoreMatchers; +import static org.hamcrest.CoreMatchers.nullValue; + +import java.io.File; +import java.io.IOException; + +/** + * @author Iwein Fuld + */ +@ContextConfiguration +@RunWith(org.springframework.test.context.junit4.SpringJUnit4ClassRunner.class) +public class FileLockingWithMultipleSourcesIntegrationTests { + private static File workdir; + + @BeforeClass + public static void setupWorkDirectory() throws Exception { + workdir = new File( + new File(System.getProperty("java.io.tmpdir")), + FileLockingWithMultipleSourcesIntegrationTests.class.getSimpleName() + ); + workdir.mkdir(); + } + + @Autowired + @Qualifier("fileSource1") + private FileReadingMessageSource fileSource1; + @Autowired + @Qualifier("fileSource2") + private FileReadingMessageSource fileSource2; + + @Before + public void cleanoutWorkDir() { + for (File file : workdir.listFiles()) { + file.delete(); + } + } + + @Test + public void filePickedUpOnlyOnce() throws IOException { + File testFile = new File(workdir, "test"); + testFile.createNewFile(); + assertThat(fileSource1.receive(), hasPayload(testFile)); + assertThat(fileSource2.receive(), nullValue()); + } +} diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/LockFileFileListFilterTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/LockFileFileListFilterTests.java new file mode 100644 index 0000000000..4ecf9af152 --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/LockFileFileListFilterTests.java @@ -0,0 +1,70 @@ +/* + * 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.locking; + +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertThat; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +/** + * @author Iwein Fuld + */ +public class LockFileFileListFilterTests { + + private File workdir = new File(new File(System.getProperty("java.io.tmpdir")), this.getClass().getSimpleName());; + private FileLocker locker = new LockFileFileLocker(workdir); + + @Before + public void setupWorkDir() { + workdir.mkdir(); + cleanDirectory(workdir); + } + + @Test + public void fileListedOnlyWhenNotLocked() throws IOException { + LockFileFileListFilter filter = new LockFileFileListFilter(workdir); + File testFile = new File(workdir, "test0"); + testFile.createNewFile(); + assertThat(filter.filterFiles(workdir.listFiles()).get(0), is(testFile)); + locker.lock(testFile); + assertThat(filter.filterFiles(workdir.listFiles()), is((List)new ArrayList())); + } + + @Test + public void fileListedByOneFilterOnly() throws IOException { + LockFileFileListFilter filter1 = new LockFileFileListFilter(workdir); + LockFileFileListFilter filter2 = new LockFileFileListFilter(workdir); + File testFile = new File(workdir, "test1"); + testFile.createNewFile(); + assertThat(filter1.filterFiles(workdir.listFiles()).get(0), is(testFile)); + locker.lock(testFile); + assertThat(filter2.filterFiles(workdir.listFiles()), is((List)new ArrayList())); + } + + private void cleanDirectory(File workdir) { + File[] files = workdir.listFiles(); + for (File file : files) { + file.delete(); + } + } + +}