diff --git a/org.springframework.integration.file/.classpath b/org.springframework.integration.file/.classpath index ae6e21a8a3..1278e1ec59 100644 --- a/org.springframework.integration.file/.classpath +++ b/org.springframework.integration.file/.classpath @@ -14,7 +14,6 @@ - diff --git a/org.springframework.integration.file/ivy.xml b/org.springframework.integration.file/ivy.xml index ffad6a5019..cd4828423d 100644 --- a/org.springframework.integration.file/ivy.xml +++ b/org.springframework.integration.file/ivy.xml @@ -22,7 +22,6 @@ - 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 99342edcbb..14d04c7e01 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 @@ -55,208 +55,212 @@ import java.util.concurrent.PriorityBlockingQueue; *

* 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 volatile FileLocker locker = new NoopFileLocker(); + private volatile FileLocker locker = new NoopFileLocker(); - private boolean scanEachPoll = false; + private boolean scanEachPoll = false; - /** - * Creates a FileReadingMessageSource with a naturally ordered queue. - */ - public FileReadingMessageSource() { - toBeReceived = new PriorityBlockingQueue(INTERNAL_QUEUE_CAPACITY); - } + /** + * Creates a FileReadingMessageSource with a naturally ordered queue. + */ + public FileReadingMessageSource() { + toBeReceived = new PriorityBlockingQueue(INTERNAL_QUEUE_CAPACITY); + } - /** - * 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 {@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 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); - } - } - } + /** + * 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); + } + } + } - /** - * 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 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; + } - /** - * 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; - } + /** + * 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; + if (filter instanceof FileLocker && locker instanceof NoopFileLocker) { + this.locker = (FileLocker) filter; + } + } - /** - * 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; - } + /** + * 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; + } - /** - * 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; - } + /** + * 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; + } - 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."); - } + 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."); + } - 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; - } + 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; + } - 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); - } - } - } + 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()); - } + /** + * 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); - } - } + /** + * 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 { + /** + * Implementation of FileLocker that doesn't provide any protection against + * duplicate listing. + */ + class NoopFileLocker implements FileLocker { - public boolean lock(File fileToLock) { - return true; - } + public boolean lock(File fileToLock) { + return true; + } - public void unlock(File fileToUnlock) { - //noop - } - } + 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 index 1ed88f9a38..b82e8b7e70 100644 --- 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 @@ -19,22 +19,25 @@ 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 to implement any relation between locking and unlocking. - * + * A FileLocker is a strategy that can ensure that files are only processed a + * single time. Implementations are free to implement any relation between + * locking and unlocking. This means that there are no safety guarantees in the + * contract, defining these guarantees is up to the implementation. + * * @author Iwein Fuld * @since 2.0 */ public interface FileLocker { - /** - * Tries to lock the given file and returns true if it was successful, false otherwise. - */ - boolean lock(File fileToLock); + /** + * 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); + /** + * Unlocks the given file. + */ + void unlock(File fileToUnlock); } diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java index 22a8c43774..4d0f591d0a 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java @@ -17,6 +17,7 @@ package org.springframework.integration.file.locking; import org.springframework.integration.file.AbstractFileListFilter; +import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.core.MessagingException; import java.io.File; @@ -26,50 +27,61 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentHashMap; /** - * File locking strategy that uses java.nio. The locks taken by FileChannel are shared with all the threads in a single - * JVM, so this locking strategy does not prevent files being picked up multiple times within the same JVM. - * + * File locking strategy that uses java.nio. The locks taken by FileChannel are + * shared with all the threads in a single JVM, so this locking strategy does + * not prevent files being picked up multiple times within the same JVM. + * {@link FileReadingMessageSource}s sharing a Locker will not pick up + * the same files. + * + * This implementation will acquire or create a {@link FileLock} for the given + * file. Caching locks might be expensive, so this locking strategy is not + * recommended for scenarios where many files are accessed in parallel. + * * @author Iwein Fuld * @author Mark Fisher * @since 2.0 */ public class NioFileLocker extends AbstractFileListFilter implements FileLocker { - private final ConcurrentMap lockCache = new ConcurrentHashMap(); + private final ConcurrentMap lockCache = new ConcurrentHashMap(); - public boolean lock(File fileToLock) { - FileLock lock = lockCache.get(fileToLock); - if (lock == null) { - FileLock newLock = null; - try { - newLock = FileChannelCache.tryLockFor(fileToLock); - } - catch (IOException e) { - throw new MessagingException("Failed to lock file: " + fileToLock, e); - } - if (newLock != null) { - FileLock original = lockCache.putIfAbsent(fileToLock, newLock); - lock = original != null ? original : newLock; - } - } - return lock != null; - } + /** + * {@inheritDoc} + * + */ + public boolean lock(File fileToLock) { + FileLock lock = lockCache.get(fileToLock); + if (lock == null) { + FileLock newLock = null; + try { + newLock = FileChannelCache.tryLockFor(fileToLock); + } catch (IOException e) { + throw new MessagingException("Failed to lock file: " + + fileToLock, e); + } + if (newLock != null) { + FileLock original = lockCache.putIfAbsent(fileToLock, newLock); + lock = original != null ? original : newLock; + } + } + return lock != null; + } - public void unlock(File fileToUnlock) { - FileLock fileLock = lockCache.get(fileToUnlock); - try { - if (fileLock != null) { - fileLock.release(); - } - FileChannelCache.closeChannelFor(fileToUnlock); - } - catch (IOException e) { - throw new MessagingException("Failed to unlock file: " + fileToUnlock, e); - } - } + public void unlock(File fileToUnlock) { + FileLock fileLock = lockCache.get(fileToUnlock); + try { + if (fileLock != null) { + fileLock.release(); + } + FileChannelCache.closeChannelFor(fileToUnlock); + } catch (IOException e) { + throw new MessagingException("Failed to unlock file: " + + fileToUnlock, e); + } + } - protected boolean accept(File file) { - return this.lock(file); - } + protected boolean accept(File file) { + return this.lock(file); + } } diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/CompositeFileListFilterTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/CompositeFileListFilterTests.java index f5d8ddc149..e3de58346b 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/CompositeFileListFilterTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/CompositeFileListFilterTests.java @@ -16,12 +16,7 @@ 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.createNiceMock; -import static org.easymock.classextension.EasyMock.replay; -import static org.easymock.classextension.EasyMock.verify; +import static org.mockito.Mockito.*; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -37,41 +32,41 @@ import org.junit.Test; */ public class CompositeFileListFilterTests { - private FileListFilter fileFilterMock1 = createMock(FileListFilter.class); + private FileListFilter fileFilterMock1 = mock(FileListFilter.class); - private FileListFilter fileFilterMock2 = createMock(FileListFilter.class); + private FileListFilter fileFilterMock2 = mock(FileListFilter.class); - private File fileMock = createNiceMock(File.class); + private File fileMock = mock(File.class); @Test public void forwardedToFilters() throws Exception { CompositeFileListFilter compositeFileFilter = new CompositeFileListFilter(fileFilterMock1, fileFilterMock2); List returnedFiles = Arrays.asList(new File[] { fileMock }); - expect(fileFilterMock1.filterFiles(isA(File[].class))).andReturn(returnedFiles).times(1); - expect(fileFilterMock2.filterFiles(isA(File[].class))).andReturn(returnedFiles).times(1); - replay(fileFilterMock1, fileFilterMock2); + when(fileFilterMock1.filterFiles(isA(File[].class))).thenReturn(returnedFiles); + when(fileFilterMock2.filterFiles(isA(File[].class))).thenReturn(returnedFiles); assertEquals(returnedFiles, compositeFileFilter.filterFiles(new File[]{fileMock})); - verify(fileFilterMock1, fileFilterMock2); + verify(fileFilterMock1).filterFiles(isA(File[].class)); + verify(fileFilterMock2).filterFiles(isA(File[].class)); } @Test public void forwardedToAddedFilters() throws Exception { CompositeFileListFilter compositeFileFilter = new CompositeFileListFilter().addFilter(fileFilterMock1, fileFilterMock2); List returnedFiles = Arrays.asList(new File[] { fileMock }); - expect(fileFilterMock1.filterFiles(isA(File[].class))).andReturn(returnedFiles).times(1); - expect(fileFilterMock2.filterFiles(isA(File[].class))).andReturn(returnedFiles).times(1); - replay(fileFilterMock1, fileFilterMock2); + when(fileFilterMock1.filterFiles(isA(File[].class))).thenReturn(returnedFiles); + when(fileFilterMock2.filterFiles(isA(File[].class))).thenReturn(returnedFiles); assertEquals(returnedFiles, compositeFileFilter.filterFiles(new File[]{fileMock})); - verify(fileFilterMock1, fileFilterMock2); + verify(fileFilterMock1).filterFiles(isA(File[].class)); + verify(fileFilterMock2).filterFiles(isA(File[].class)); } @Test public void negative() throws Exception { CompositeFileListFilter compositeFileFilter = new CompositeFileListFilter(fileFilterMock1, fileFilterMock2); - expect(fileFilterMock2.filterFiles(isA(File[].class))).andReturn(new ArrayList()).times(1); - expect(fileFilterMock1.filterFiles(isA(File[].class))).andReturn(new ArrayList()).times(1); - replay(fileFilterMock1, fileFilterMock2); + when(fileFilterMock2.filterFiles(isA(File[].class))).thenReturn(new ArrayList()); + when(fileFilterMock1.filterFiles(isA(File[].class))).thenReturn(new ArrayList()); assertTrue(compositeFileFilter.filterFiles(new File[]{fileMock}).isEmpty()); - verify(fileFilterMock1, fileFilterMock2); + verify(fileFilterMock1).filterFiles(isA(File[].class)); + verify(fileFilterMock2).filterFiles(isA(File[].class)); } } 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 index 850f4e20b3..e7acd0332c 100644 --- 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 @@ -8,10 +8,10 @@ + p:filter-ref="filter"/> + p:filter-ref="filter"/>