put threaded tests and ThreadLocal buffering place for FtpSource, FileSource should work as before, but there are no test for it yet.

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java

@@ -63,10 +63,9 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 	@SuppressWarnings("unchecked")
 	public final Message<T> receive() {
 		try {
-			HashMap<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
-			this.populateSnapshot(snapshot);
-			this.directoryContentManager.processSnapshot(snapshot);
-			if (!getDirectoryContentManager().getBacklog().isEmpty()) {
+			refreshSnapshotAndMarkProcessing(directoryContentManager);
+			if (!(getDirectoryContentManager().getProcessingBuffer().isEmpty() & getDirectoryContentManager()
+					.getBacklog().isEmpty())) {
 				return buildNextMessage();
 			}
 			return null;
@@ -76,6 +75,20 @@ 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 DirectoryContentManager} override this method and call
+	 * <code>directoryContentManager.fileProcessing(...)</code with the appropriate arguments
+	 * @param directoryContentManager
+	 * @throws IOException
+	 */
+	protected void refreshSnapshotAndMarkProcessing(DirectoryContentManager directoryContentManager) throws IOException {
+		HashMap<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
+		this.populateSnapshot(snapshot);
+		directoryContentManager.processSnapshot(snapshot);
+	}
+
 	/**
 	 * Hook point for implementors to create the next message that should be
 	 * received. Implementations can use a File by File approach like
@@ -113,8 +126,8 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 	 */
 	protected abstract T retrieveNextPayload() throws IOException;
 
-	protected final void fileProcessed(String fileName){
+	protected final void fileProcessed(String fileName) {
 		this.directoryContentManager.fileProcessed(fileName);
 	}
-	
+
 }

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java

@@ -20,9 +20,11 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.util.Assert;
 
 /**
  * Tracks changes in a directory. This implementation is thread-safe as it
@@ -30,6 +32,7 @@ import org.apache.commons.logging.LogFactory;
  * 
  * @author Marius Bogoevici
  * @author Mark Fisher
+ * @author iwein
  */
 public class DirectoryContentManager {
 
@@ -39,14 +42,33 @@ public class DirectoryContentManager {
 
 	private final Map<String, FileInfo> backlog = new HashMap<String, FileInfo>();
 
+	/**
+	 * 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)
+	 */
+	private ThreadLocal<Map<String, FileInfo>> processingBuffer = new ThreadLocal<Map<String, FileInfo>>() {
+		@Override
+		protected Map<String, FileInfo> initialValue() {
+			return new HashMap<String, FileInfo>();
+		}
+	};
 
 	public synchronized void processSnapshot(Map<String, FileInfo> currentSnapshot) {
+		/*
+		 * clear the threadLocal backlog. When the thread processes a new
+		 * snapshot it is done with the previous message. If there are still
+		 * messages in the processing buffer something is wrong because they
+		 * were not processed, nor raised as failed.
+		 */
+		Assert.isTrue(processingBuffer.get().isEmpty());
 		Iterator<Map.Entry<String, FileInfo>> iter = this.backlog.entrySet().iterator();
 		while (iter.hasNext()) {
 			String fileName = iter.next().getKey();
 			if (!currentSnapshot.containsKey(fileName)) {
 				if (logger.isDebugEnabled()) {
-					logger.debug("Removing file '" + fileName + "' from backlog. It no longer exists in remote directory.");
+					logger.debug("Removing file '" + fileName
+							+ "' from backlog. It no longer exists in remote directory.");
 				}
 				iter.remove();
 			}
@@ -63,12 +85,42 @@ public class DirectoryContentManager {
 		this.previousSnapshot = new HashMap<String, FileInfo>(currentSnapshot);
 	}
 
-	public synchronized void fileProcessed(String fileName) {
-		if (fileName != null) {
-			if (logger.isDebugEnabled()) {
-				logger.debug("Removing file '" + fileName + "' from the backlog. It has been processed.");
+	public synchronized void fileProcessing(String... fileNames) {
+		for (String fileName : fileNames) {
+			if (fileName != null) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("Moving file '" + fileName
+							+ "' from the backlog to thread local backlog. It is being processed.");
+				}
+				processingBuffer.get().put(fileName, this.backlog.remove(fileName));
+			}
+		}
+	}
+
+	public synchronized void fileNotProcessed(String... fileNames) {
+		for (String fileName : fileNames) {
+			if (fileName != null) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("File '" + fileName + "' was not processed. Moving it back to the backlog");
+				}
+				this.backlog.put(fileName, this.processingBuffer.get().remove(fileName));
+			}
+		}
+	}
+
+	public synchronized void fileProcessed(String... fileNames) {
+		for (String fileName : fileNames) {
+			if (fileName != null) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("Removing file '" + fileName + "' from the undo buffer. It has been processed.");
+				}
+				/*
+				 * It's not relevant if clients use the thread safe approach or
+				 * not at this point, so we act on both.
+				 */
+				this.processingBuffer.get().remove(fileName);
+				this.backlog.remove(fileName);
 			}
-			this.backlog.remove(fileName);
 		}
 	}
 
@@ -76,4 +128,8 @@ public class DirectoryContentManager {
 		return Collections.unmodifiableMap(this.backlog);
 	}
 
+	public Map<String, FileInfo> getProcessingBuffer() {
+		return Collections.unmodifiableMap(this.processingBuffer.get());
+	}
+
 }

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java

@@ -20,6 +20,8 @@ import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -33,6 +35,7 @@ import org.apache.commons.net.ftp.FTPFile;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.annotation.Required;
 import org.springframework.integration.adapter.file.AbstractDirectorySource;
+import org.springframework.integration.adapter.file.DirectoryContentManager;
 import org.springframework.integration.adapter.file.FileInfo;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageCreator;
@@ -115,15 +118,33 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
 		this.localWorkingDirectory = localWorkingDirectory;
 	}
 
+	@Override
+	protected void refreshSnapshotAndMarkProcessing(DirectoryContentManager directoryContentManager) throws IOException {
+		Map<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
+		synchronized (directoryContentManager) {
+			populateSnapshot(snapshot);
+			directoryContentManager.processSnapshot(snapshot);
+			ArrayList<String> backlog = new ArrayList<String>(directoryContentManager.getBacklog().keySet());
+			int toIndex = maxFilesPerPayload == -1 ? backlog.size() : Math.min(maxFilesPerPayload, backlog.size());
+			directoryContentManager.fileProcessing(backlog.subList(0, toIndex).toArray(new String[] {}));
+		}
+	}
+
 	@Override
 	protected void populateSnapshot(Map<String, FileInfo> snapshot) throws IOException {
 		establishConnection();
 		FTPFile[] fileList = this.client.listFiles();
 
 		for (FTPFile ftpFile : fileList) {
-			FileInfo fileInfo = new FileInfo(ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(), ftpFile
-					.getSize());
-			snapshot.put(ftpFile.getName(), fileInfo);
+			/*
+			 * according to the FTPFile javadoc the list can contain nulls if
+			 * files couldn't be parsed
+			 */
+			if (ftpFile != null) {
+				FileInfo fileInfo = new FileInfo(ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(), ftpFile
+						.getSize());
+				snapshot.put(ftpFile.getName(), fileInfo);
+			}
 		}
 	}
 
@@ -158,13 +179,8 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
 	protected List<File> retrieveNextPayload() throws IOException {
 		establishConnection();
 		List<File> files = new ArrayList<File>();
-		Set<String> backlog = this.getDirectoryContentManager().getBacklog().keySet();
-		int i = 0;
-		Iterator<String> iterator = backlog.iterator();
-		while (iterator.hasNext() && (maxFilesPerPayload == -1 || i < maxFilesPerPayload)) {
-			i++;
-			String fileName = iterator.next();
-
+		Set<String> toDo = this.getDirectoryContentManager().getProcessingBuffer().keySet();
+		for (String fileName : toDo) {
 			File file = new File(this.localWorkingDirectory, fileName);
 			if (file.exists()) {
 				file.delete();