Complete: Test Backlog concurrency

Incomplete: Improve FileSource. Added comparator constructor arg. Still need namespace cleanup

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<T> implements PollableSource<T>, M
 
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
-	private final Backlog<FileSnapshot> directoryContentManager = new Backlog<FileSnapshot>();
+	private final Backlog<FileSnapshot> backlog;
 
 	private final MessageCreator<T, T> messageCreator;
 
 	public AbstractDirectorySource(MessageCreator<T, T> messageCreator) {
+		this(messageCreator, null);
+	}
+
+	public AbstractDirectorySource(MessageCreator<T, T> messageCreator, Comparator<FileSnapshot> comparator) {
+		this.backlog = comparator == null ? new Backlog<FileSnapshot>() : new Backlog<FileSnapshot>(comparator);
 		Assert.notNull(messageCreator, "The MessageCreator must not be null");
 		this.messageCreator = messageCreator;
 	}
 
 	protected Backlog<FileSnapshot> getBacklog() {
-		return this.directoryContentManager;
+		return this.backlog;
 	}
 
 	public MessageCreator<T, T> getMessageCreator() {
@@ -61,7 +67,7 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 
 	public final Message<T> receive() {
 		try {
-			refreshSnapshotAndMarkProcessing(this.directoryContentManager);
+			refreshSnapshotAndMarkProcessing(this.backlog);
 			if (!getBacklog().isEmpty()) {
 				return buildNextMessage();
 			}
@@ -72,20 +78,10 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, 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
-	 * <code>directoryContentManager.fileProcessing(...)</code> with the
-	 * appropriate arguments.
-	 * 
-	 * @param directoryContentManager
-	 * @throws IOException
-	 */
-	protected void refreshSnapshotAndMarkProcessing(Backlog<FileSnapshot> directoryContentManager) throws IOException {
+	protected void refreshSnapshotAndMarkProcessing(Backlog<FileSnapshot> backlog) throws IOException {
 		List<FileSnapshot> snapshot = new ArrayList<FileSnapshot>();
 		this.populateSnapshot(snapshot);
-		directoryContentManager.processSnapshot(snapshot);
+		backlog.processSnapshot(snapshot);
 	}
 
 	/**
@@ -107,7 +103,7 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, 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<T> implements PollableSource<T>, 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<T> implements PollableSource<T>, M
 	protected abstract T retrieveNextPayload() throws IOException;
 
 	protected final void filesProcessed() {
-		this.directoryContentManager.processed();
+		this.backlog.processed();
 	}
 }

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<T extends Comparable<T>> {
 
+	private static final int INITIAL_QUEUE_CAPACITY = 5;
+
 	private final Log logger = LogFactory.getLog(this.getClass());
 
-	private PriorityBlockingQueue<T> backlog = new PriorityBlockingQueue<T>();
+	/*
+	 * 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<T> backlog;
 
-	/**
+	// @GuardedBy(this)
+	private Set<T> doneProcessing = new HashSet<T>();
+
+	// @GuardedBy(this)
+	private Set<T> currentlyProcessing = new HashSet<T>();
+
+	/*
 	 * 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<List<T>> processingBuffer = new ThreadLocal<List<T>>() {
 		@Override
@@ -53,11 +69,25 @@ public class Backlog<T extends Comparable<T>> {
 		}
 	};
 
-	private Set<T> doneProcessing = Collections.synchronizedSet(new HashSet<T>());
+	/**
+	 * Constructs a Backlog around a naturally ordered
+	 * {@link PriorityBlockingQueue}.
+	 */
+	public Backlog() {
+		this.backlog = new PriorityBlockingQueue<T>();
+	}
 
-	private Set<T> currentlyProcessing = Collections.synchronizedSet(new HashSet<T>());
+	/**
+	 * 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<? super T> comparator) {
+		this.backlog = new PriorityBlockingQueue<T>(INITIAL_QUEUE_CAPACITY, comparator);
+	}
 
-	public synchronized void processSnapshot(List<T> currentSnapshot) {
+	public void processSnapshot(List<T> 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<T extends Comparable<T>> {
 		 * 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<T> newInCurrentSnapshot = new ArrayList<T>(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<T extends Comparable<T>> {
 	 * this method can be used to mark a subset of the processing buffer as
 	 * processed. Calling this method in a threaded scenario without using
 	 * <code>{@link #prepareForProcessing(int)}</code> 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<T extends Comparable<T>> {
 		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<T extends Comparable<T>> {
 			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<T> processing = this.processingBuffer.get();
-		this.backlog.addAll(processing);
-		currentlyProcessing.removeAll(processing);
+		synchronized (this) {
+			this.backlog.addAll(processing);
+			currentlyProcessing.removeAll(processing);
+		}
 		processing.clear();
 	}
 

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<File> 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<File> buildNextMessage() throws IOException {
-		File file = retrieveNextPayload();
-		Message<File> 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<FileSnapshot> comparator) {
+		this(directory, new FileMessageCreator(), comparator);
 	}
 
 	public FileSource(Resource directory, MessageCreator<File, File> 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
+	 * <code>null</code> natural order is used.
+	 */
+	public FileSource(Resource directory, MessageCreator<File, File> messageCreator, Comparator<FileSnapshot> 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<File> implements Message
 		}
 	}
 
+	@Override
+	protected Message<File> buildNextMessage() throws IOException {
+		File file = retrieveNextPayload();
+		Message<File> 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
 	 * <code>{@link File#listFiles()}</code> 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<File> implements Message
 	@Override
 	protected File retrieveNextPayload() throws IOException {
 		List<FileSnapshot> selectedForProcessing = this.getBacklog().selectForProcessing(1);
-		if(!selectedForProcessing.isEmpty()){
+		if (!selectedForProcessing.isEmpty()) {
 			return selectedForProcessing.get(0).getFile();
-		} else{
+		}
+		else {
 			return null;
 		}
 	}

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<String> items = Arrays.asList(new String[] { "bert", "ernie", "pino", "whatsherface" });
+		List<String> 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)

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<List<File>> received = ftpSource.receive();