Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

IN PROGRESS - issue INT-472: Concurrency issue in FileReadingMessageSource

Browse Source 
http://jira.springframework.org/browse/INT-472

removed removeAll call to prevent iteration over the 'toBeDelivered' queue.
This commit is contained in:
Iwein Fuld
2008-11-16 10:34:37 +00:00
parent 19b4557d63
commit 304afa1e38
1 changed files with 23 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -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.
* <p>
* default, an {@link AcceptOnceFileListFilter} is used. It ensures files are
* picked up only once from the directory.
* <p/>
* 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.
* <p/>
* FileReadingMessageSource is fully thread-safe under concurrent receives and
* message delivery callbacks.
*
* @author Iwein Fuld
*/
@@ -58,11 +60,15 @@ public class FileReadingMessageSource implements MessageSource<File>, 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<File> toBeReceived = new PriorityBlockingQueue<File>();
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<File>, 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 <p/>
* {@link CompositeFileListFilter} can be used to group them together
* <p/>
* <b>Note that the supplied filter must be thread safe</b>.
*/
public void setFilter(FileListFilter filter) {
@@ -106,8 +113,6 @@ public class FileReadingMessageSource implements MessageSource<File>, MessageDel
List<File> filteredFiles = filter.filterFiles((inputDirectory.listFiles()));
Set<File> freshFiles = new HashSet<File>(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<File>, 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<File> 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 <code>receive()</code>
*/
public void onSend(Message<File> sentMessage) {
if (logger.isDebugEnabled()) {
logger.debug("Sent: " + sentMessage);