From 304afa1e388760555b0a15198971527487150977 Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Sun, 16 Nov 2008 10:34:37 +0000 Subject: [PATCH] IN PROGRESS - issue INT-472: Concurrency issue in FileReadingMessageSource http://jira.springframework.org/browse/INT-472 removed removeAll call to prevent iteration over the 'toBeDelivered' queue. --- .../file/FileReadingMessageSource.java | 39 +++++++++++-------- 1 file changed, 23 insertions(+), 16 deletions(-) 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 e0f38ab5c0..13a7e966fd 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 @@ -27,7 +27,6 @@ import java.util.concurrent.PriorityBlockingQueue; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.core.io.Resource; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessagingException; @@ -39,16 +38,19 @@ import org.springframework.util.Assert; /** * PollableSource that creates messages from a file system directory. To prevent * messages for certain files, you may supply a {@link FileListFilter}. By - * default, an {@link AcceptOnceFileListFilter} is used. It ensures files are picked - * up only once from the directory. - *

+ * default, an {@link AcceptOnceFileListFilter} is used. It ensures files are + * picked up only once from the directory. + *

* A common problem with reading files is that a file may be detected before it - * is ready. The default {@link AcceptOnceFileListFilter} does not prevent this. In - * most cases, this can be prevented if the file-writing process renames each + * is ready. The default {@link AcceptOnceFileListFilter} does not prevent this. + * In most cases, this can be prevented if the file-writing process renames each * file as soon as it is ready for reading. A pattern-matching filter that * accepts only files that are ready (e.g. based on a known suffix), composed - * with the default {@link AcceptOnceFileListFilter} would allow for this. - * See {@ link CompositeFileFilter} for a way to do this. + * with the default {@link AcceptOnceFileListFilter} would allow for this. See + * {@link CompositeFileFilter} for a way to do this. + *

+ * FileReadingMessageSource is fully thread-safe under concurrent receives and + * message delivery callbacks. * * @author Iwein Fuld */ @@ -58,11 +60,15 @@ public class FileReadingMessageSource implements MessageSource, MessageDel private volatile File inputDirectory; + /** + * {@link PriorityBlockingQueue#iterator()} throws + * {@link ConcurrentModificationException} in Java 5. There is no locking + * around the queue, so there is also no iteration. + */ private final Queue toBeReceived = new PriorityBlockingQueue(); private volatile FileListFilter filter = new AcceptOnceFileListFilter(); - public void setInputDirectory(Resource inputDirectory) { Assert.notNull(inputDirectory, "inputDirectory cannot be null"); Assert.isTrue(inputDirectory.exists(), inputDirectory + " doesn't exist."); @@ -80,7 +86,8 @@ public class FileReadingMessageSource implements MessageSource, MessageDel * {@link AcceptOnceFileListFilter} with no bounds is used. In most cases a * customized {@link FileFilter} 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

+ * {@link CompositeFileListFilter} can be used to group them together + *

* Note that the supplied filter must be thread safe. */ public void setFilter(FileListFilter filter) { @@ -106,8 +113,6 @@ public class FileReadingMessageSource implements MessageSource, MessageDel List filteredFiles = filter.filterFiles((inputDirectory.listFiles())); Set freshFiles = new HashSet(filteredFiles); if (!freshFiles.isEmpty()) { - // don't duplicate what's on the queue already - freshFiles.removeAll(toBeReceived); toBeReceived.addAll(freshFiles); if (logger.isDebugEnabled()) { logger.debug("Added to queue: " + freshFiles); @@ -116,17 +121,19 @@ public class FileReadingMessageSource implements MessageSource, MessageDel } /** - * In concurrent scenarios onFailure() might cause failing files to be - * ignored. If this is not acceptable access to this method should be - * synchronized on this instance externally. + * 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.add(failedMessage.getPayload()); + 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);