diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java index 9e9f4adf7f..65380adc4c 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java @@ -18,6 +18,7 @@ package org.springframework.integration.adapter.file; import java.io.IOException; import java.util.ArrayList; +import java.util.Comparator; import java.util.List; import org.apache.commons.logging.Log; @@ -42,17 +43,22 @@ public abstract class AbstractDirectorySource implements PollableSource, M protected final Log logger = LogFactory.getLog(this.getClass()); - private final Backlog directoryContentManager = new Backlog(); + private final Backlog backlog; private final MessageCreator messageCreator; public AbstractDirectorySource(MessageCreator messageCreator) { + this(messageCreator, null); + } + + public AbstractDirectorySource(MessageCreator messageCreator, Comparator comparator) { + this.backlog = comparator == null ? new Backlog() : new Backlog(comparator); Assert.notNull(messageCreator, "The MessageCreator must not be null"); this.messageCreator = messageCreator; } protected Backlog getBacklog() { - return this.directoryContentManager; + return this.backlog; } public MessageCreator getMessageCreator() { @@ -61,7 +67,7 @@ public abstract class AbstractDirectorySource implements PollableSource, M public final Message receive() { try { - refreshSnapshotAndMarkProcessing(this.directoryContentManager); + refreshSnapshotAndMarkProcessing(this.backlog); if (!getBacklog().isEmpty()) { return buildNextMessage(); } @@ -72,20 +78,10 @@ public abstract class AbstractDirectorySource implements PollableSource, M } } - /** - * Naive implementation that ignores thread safety. Subclasses that want to - * be thread safe and use the reservation facilities of {@link Backlog} - * should override this method and call - * directoryContentManager.fileProcessing(...) with the - * appropriate arguments. - * - * @param directoryContentManager - * @throws IOException - */ - protected void refreshSnapshotAndMarkProcessing(Backlog directoryContentManager) throws IOException { + protected void refreshSnapshotAndMarkProcessing(Backlog backlog) throws IOException { List snapshot = new ArrayList(); this.populateSnapshot(snapshot); - directoryContentManager.processSnapshot(snapshot); + backlog.processSnapshot(snapshot); } /** @@ -107,7 +103,7 @@ public abstract class AbstractDirectorySource implements PollableSource, M if (logger.isDebugEnabled()) { logger.debug(message + " processed successfully. Files will be removed from backlog"); } - this.directoryContentManager.processed(); + this.backlog.processed(); } public void onFailure(Message failedMessage, Throwable exception) { @@ -115,7 +111,7 @@ public abstract class AbstractDirectorySource implements PollableSource, M this.logger.warn("Failure notification received by [" + this.getClass().getSimpleName() + "] for message: " + failedMessage + ". Selected files will be moved back to the backlog.", exception); } - this.directoryContentManager.processingFailed(); + this.backlog.processingFailed(); } /** @@ -135,6 +131,6 @@ public abstract class AbstractDirectorySource implements PollableSource, M protected abstract T retrieveNextPayload() throws IOException; protected final void filesProcessed() { - this.directoryContentManager.processed(); + this.backlog.processed(); } } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java index 9dcde18b80..5882b76230 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java @@ -19,6 +19,7 @@ package org.springframework.integration.adapter.file; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -37,14 +38,29 @@ import org.springframework.util.Assert; */ public class Backlog> { + private static final int INITIAL_QUEUE_CAPACITY = 5; + private final Log logger = LogFactory.getLog(this.getClass()); - private PriorityBlockingQueue backlog = new PriorityBlockingQueue(); + /* + * Backlog, doneProcessing and currentlyProcessing should be in consistent + * state together. To do that access to them is synchronized on this and + * atomic operations have been defined on this class. It is not a problem if + * items exist in more than one of these collections at the same time, but the + * item should not be removed from one collection before it is added to the + * next. + */ + private final PriorityBlockingQueue backlog; - /** + // @GuardedBy(this) + private Set doneProcessing = new HashSet(); + + // @GuardedBy(this) + private Set currentlyProcessing = new HashSet(); + + /* * This is the storage for backlog that is being processed by a specific - * thread. Not initialized means that we're not in thread safe mode (just - * working directly on the backlog) + * thread. */ private ThreadLocal> processingBuffer = new ThreadLocal>() { @Override @@ -53,11 +69,25 @@ public class Backlog> { } }; - private Set doneProcessing = Collections.synchronizedSet(new HashSet()); + /** + * Constructs a Backlog around a naturally ordered + * {@link PriorityBlockingQueue}. + */ + public Backlog() { + this.backlog = new PriorityBlockingQueue(); + } - private Set currentlyProcessing = Collections.synchronizedSet(new HashSet()); + /** + * Constructs a backlog around a {@link PriorityBlockingQueue} that is + * created with the supplied comparator. For natural ordering use + * {@link #Backlog()}. + * @param comparator + */ + public Backlog(Comparator comparator) { + this.backlog = new PriorityBlockingQueue(INITIAL_QUEUE_CAPACITY, comparator); + } - public synchronized void processSnapshot(List currentSnapshot) { + public void processSnapshot(List currentSnapshot) { /* * clear the threadLocal backlog. When the thread processes a new * snapshot it is done with the previous message. If there are still @@ -65,16 +95,20 @@ public class Backlog> { * were not processed, nor raised as failed. */ Assert.isTrue(processingBuffer.get().isEmpty(), "Processing buffer not emptied before poll."); - // remove everything that is not in the latest snapshot from the backlog - backlog.retainAll(currentSnapshot); - doneProcessing.retainAll(currentSnapshot); - // add everything that is new to the backlog preventing side effect on - // currentSnapshot + /* + * compute everything that is new to the backlog preventing side effect + * on currentSnapshot. + */ Collection newInCurrentSnapshot = new ArrayList(currentSnapshot); - newInCurrentSnapshot.removeAll(backlog); - newInCurrentSnapshot.removeAll(doneProcessing); - newInCurrentSnapshot.removeAll(currentlyProcessing); - backlog.addAll(newInCurrentSnapshot); + synchronized (this) { + newInCurrentSnapshot.removeAll(backlog); + newInCurrentSnapshot.removeAll(currentlyProcessing); + newInCurrentSnapshot.removeAll(doneProcessing); + + backlog.retainAll(currentSnapshot); + doneProcessing.retainAll(currentSnapshot); + backlog.addAll(newInCurrentSnapshot); + } } /** @@ -84,20 +118,20 @@ public class Backlog> { * this method can be used to mark a subset of the processing buffer as * processed. Calling this method in a threaded scenario without using * {@link #prepareForProcessing(int)} will manipulate the - * backlog directly and allows for race conditions. In threaded scenarios a - * call to {@link #selectForProcessing(int)} followed by a call to + * backlog directly. In threaded scenarios a call to + * {@link #selectForProcessing(int)} followed by a call to * {@link #processed()} is recommended. * @param items the items that have been processed */ - public void fileProcessed(T... items) { + public synchronized void fileProcessed(T... items) { for (T item : items) { if (item != null) { if (logger.isDebugEnabled()) { logger.debug("Removing item '" + item + "' from the undo buffer. It has been processed."); } - this.processingBuffer.get().remove(item); - this.backlog.remove(item); this.doneProcessing.add(item); + this.backlog.remove(item); + this.processingBuffer.get().remove(item); } } } @@ -151,14 +185,10 @@ public class Backlog> { if (logger.isDebugEnabled()) { logger.debug("Moving processing buffer " + processingBuffer.get() + " to doneProcessing"); } - /* - * this doesn't need synchronization because the processing buffer will - * first be moved to doneProcessing before it is removed from currently - * processing. The order and the thread safety of the used collections - * is essential. - */ - this.doneProcessing.addAll(this.processingBuffer.get()); - currentlyProcessing.removeAll(processingBuffer.get()); + synchronized (this) { + this.doneProcessing.addAll(this.processingBuffer.get()); + currentlyProcessing.removeAll(processingBuffer.get()); + } this.processingBuffer.get().clear(); } @@ -171,15 +201,11 @@ public class Backlog> { logger.debug("Moving processing buffer " + processingBuffer.get() + " back to backlog. Processing has failed"); } - /* - * this doesn't need synchronization because the processing buffer will - * first be moved to doneProcessing before it is removed from currently - * processing. The order and the thread safety of the used collections - * is essential. - */ List processing = this.processingBuffer.get(); - this.backlog.addAll(processing); - currentlyProcessing.removeAll(processing); + synchronized (this) { + this.backlog.addAll(processing); + currentlyProcessing.removeAll(processing); + } processing.clear(); } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java index 7a1fdecfa3..bc0a40120c 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java @@ -20,6 +20,7 @@ import java.io.File; import java.io.FileFilter; import java.io.FilenameFilter; import java.io.IOException; +import java.util.Comparator; import java.util.List; import org.springframework.core.io.Resource; @@ -46,22 +47,32 @@ public class FileSource extends AbstractDirectorySource implements Message private volatile FilenameFilter filenameFilter; + /** + * Creates a default FileSource on the specified directory + * @param directory + */ public FileSource(Resource directory) { - this(directory, new FileMessageCreator()); + this(directory, new FileMessageCreator(), null); } - @Override - protected Message buildNextMessage() throws IOException { - File file = retrieveNextPayload(); - Message message = this.getMessageCreator().createMessage(file); - message = MessageBuilder.fromMessage(message) - .setHeader(FileNameGenerator.FILENAME_PROPERTY_KEY, file.getName()).setHeader(FILE_INFO_PROPERTY, - getBacklog().getProcessingBuffer().get(0)).build(); - return message; + public FileSource(Resource directory, Comparator comparator) { + this(directory, new FileMessageCreator(), comparator); } public FileSource(Resource directory, MessageCreator messageCreator) { - super(messageCreator); + this(directory, messageCreator, null); + } + + /** + * Creates a FileSource with the specified strategies. + * @param directory + * @param messageCreator the MessageCreator used to convert Files into + * Messages + * @param comparator the comparator is used to order the backlog. If + * null natural order is used. + */ + public FileSource(Resource directory, MessageCreator messageCreator, Comparator comparator) { + super(messageCreator, comparator); Assert.notNull(directory, "The directory must not be null"); try { this.directory = directory.getFile(); @@ -75,11 +86,21 @@ public class FileSource extends AbstractDirectorySource implements Message } } + @Override + protected Message buildNextMessage() throws IOException { + File file = retrieveNextPayload(); + Message message = this.getMessageCreator().createMessage(file); + message = MessageBuilder.fromMessage(message) + .setHeader(FileNameGenerator.FILENAME_PROPERTY_KEY, file.getName()).setHeader(FILE_INFO_PROPERTY, + getBacklog().getProcessingBuffer().get(0)).build(); + return message; + } + /** * Sets a FilenameFilter to be used with the * {@link File#listFiles()} command. Note that either a * FileFilter or a FilenameFilter is used to filter the list of files. - * Calling this setter overwrites the FileFilter if it was set before. + * Calling this setter overwrites the FileNameFilter if it was set before. * @param fileFilter */ public void setFileFilter(FileFilter fileFilter) { @@ -126,9 +147,10 @@ public class FileSource extends AbstractDirectorySource implements Message @Override protected File retrieveNextPayload() throws IOException { List selectedForProcessing = this.getBacklog().selectForProcessing(1); - if(!selectedForProcessing.isEmpty()){ + if (!selectedForProcessing.isEmpty()) { return selectedForProcessing.get(0).getFile(); - } else{ + } + else { return null; } } diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/ConcurrentBacklogTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/ConcurrentBacklogTests.java index 9e78b8d2bd..eaa31acbca 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/ConcurrentBacklogTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/ConcurrentBacklogTests.java @@ -26,7 +26,12 @@ public class ConcurrentBacklogTests { CountDownLatch start = new CountDownLatch(1); CountDownLatch done = doConcurrently(5, todo, start); start.countDown(); - done.await(); + try { + done.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } assertTrue(backlog.isEmpty()); } @@ -46,7 +51,12 @@ public class ConcurrentBacklogTests { CountDownLatch start = new CountDownLatch(1); CountDownLatch done = doConcurrently(3, todo, start); start.countDown(); - done.await(); + try { + done.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } assertTrue("currentlyProcessing not emptied correctly", ((Collection) new DirectFieldAccessor(backlog) .getPropertyValue("currentlyProcessing")).isEmpty()); assertTrue("doneProcessing not populated correctly", ((Collection) new DirectFieldAccessor(backlog) @@ -69,7 +79,12 @@ public class ConcurrentBacklogTests { CountDownLatch start = new CountDownLatch(1); CountDownLatch done = doConcurrently(3, todo, start); start.countDown(); - done.await(); + try { + done.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } assertTrue("currentlyProcessing not emptied correctly", ((Collection) new DirectFieldAccessor(backlog) .getPropertyValue("currentlyProcessing")).isEmpty()); assertTrue("backlog not repopulated correctly", ((Collection) new DirectFieldAccessor(backlog) @@ -79,7 +94,7 @@ public class ConcurrentBacklogTests { @Test(timeout = 1000) public void concurrentSuccessFailure() throws Exception { final Backlog backlog = new Backlog(); - List items = Arrays.asList(new String[] { "bert", "ernie", "pino", "whatsherface" }); + List items = Arrays.asList(new String[] { "ham", "chicken", "burger", "cheeze" }); backlog.processSnapshot(items); final AtomicBoolean properlyBackedUp = new AtomicBoolean(true); final AtomicBoolean properlyUnloaded = new AtomicBoolean(true); @@ -93,16 +108,26 @@ public class ConcurrentBacklogTests { Runnable doSuccess = new Runnable() { public void run() { backlog.prepareForProcessing(1); + //make sure we process a message + while (backlog.getProcessingBuffer().size() == 0) { + Thread.yield(); + backlog.prepareForProcessing(1); + } backlog.processed(); properlyUnloaded.set(backlog.getProcessingBuffer().isEmpty() && properlyUnloaded.get()); } }; CountDownLatch start = new CountDownLatch(1); + CountDownLatch doneFailure = doConcurrently(20, doFailure, start); CountDownLatch doneSuccess = doConcurrently(2, doSuccess, start); - CountDownLatch doneFailure = doConcurrently(3, doFailure, start); start.countDown(); - doneSuccess.await(); - doneFailure.await(); + try { + doneSuccess.await(); + doneFailure.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } assertTrue(properlyBackedUp.get()); assertTrue(properlyUnloaded.get()); assertTrue("currentlyProcessing not emptied correctly", ((Collection) new DirectFieldAccessor(backlog) diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceIntegrationTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceIntegrationTests.java index 23f94a83ed..ef0a5c9ca7 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceIntegrationTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceIntegrationTests.java @@ -67,7 +67,6 @@ public class FtpSourceIntegrationTests { queuedFTPClientPool.setRemoteWorkingDirectory("ftp-test"); } - @SuppressWarnings("unchecked") @Test public void receive() { Message> received = ftpSource.receive();