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 69c2d978be..5f516e0624 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 @@ -63,10 +63,9 @@ public abstract class AbstractDirectorySource implements PollableSource, M @SuppressWarnings("unchecked") public final Message receive() { try { - HashMap snapshot = new HashMap(); - 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 implements PollableSource, 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 + * directoryContentManager.fileProcessing(...) snapshot = new HashMap(); + 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 implements PollableSource, M */ protected abstract T retrieveNextPayload() throws IOException; - protected final void fileProcessed(String fileName){ + protected final void fileProcessed(String fileName) { this.directoryContentManager.fileProcessed(fileName); } - + } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java index ea5e7ce3a5..457f9cbdb9 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java +++ b/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 backlog = new HashMap(); + /** + * 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> processingBuffer = new ThreadLocal>() { + @Override + protected Map initialValue() { + return new HashMap(); + } + }; public synchronized void processSnapshot(Map 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> 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(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 getProcessingBuffer() { + return Collections.unmodifiableMap(this.processingBuffer.get()); + } + } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java index d8ea345797..90bfed9829 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java +++ b/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> implements Di this.localWorkingDirectory = localWorkingDirectory; } + @Override + protected void refreshSnapshotAndMarkProcessing(DirectoryContentManager directoryContentManager) throws IOException { + Map snapshot = new HashMap(); + synchronized (directoryContentManager) { + populateSnapshot(snapshot); + directoryContentManager.processSnapshot(snapshot); + ArrayList backlog = new ArrayList(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 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> implements Di protected List retrieveNextPayload() throws IOException { establishConnection(); List files = new ArrayList(); - Set backlog = this.getDirectoryContentManager().getBacklog().keySet(); - int i = 0; - Iterator iterator = backlog.iterator(); - while (iterator.hasNext() && (maxFilesPerPayload == -1 || i < maxFilesPerPayload)) { - i++; - String fileName = iterator.next(); - + Set toDo = this.getDirectoryContentManager().getProcessingBuffer().keySet(); + for (String fileName : toDo) { File file = new File(this.localWorkingDirectory, fileName); if (file.exists()) { file.delete(); diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java index 313719110f..37fb864e76 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java @@ -16,7 +16,7 @@ package org.springframework.integration.adapter.ftp; -import static org.easymock.EasyMock.anyInt; +import static org.easymock.EasyMock.*; import static org.easymock.EasyMock.eq; import static org.easymock.EasyMock.expect; import static org.easymock.EasyMock.getCurrentArguments; @@ -25,6 +25,7 @@ import static org.easymock.classextension.EasyMock.createMock; import static org.easymock.classextension.EasyMock.replay; import static org.easymock.classextension.EasyMock.reset; import static org.easymock.classextension.EasyMock.verify; +import static org.junit.Assert.*; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; @@ -35,6 +36,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Calendar; import java.util.List; +import java.util.concurrent.CountDownLatch; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; @@ -67,7 +69,7 @@ public class FtpSourceTests { private FtpSource ftpSource; private Long size = 100l; - + @Before public void initializeFtpSource() { ftpSource = new FtpSource(messageCreator, ftpClient); @@ -98,7 +100,8 @@ public class FtpSourceTests { new GenericMessage(Arrays.asList(new File("test1")))); ftpClient.disconnect(); replay(globalMocks); - ftpSource.onSend(ftpSource.receive()); + Message> received = ftpSource.receive(); + ftpSource.onSend(received); verify(globalMocks); } @@ -182,19 +185,22 @@ public class FtpSourceTests { @Test public void retrieveMaxFilesPerPayload() throws Exception { - this.ftpSource.setMaxMessagesPerPayload(2); + this.ftpSource.setMaxMessagesPerPayload(2); // assume client already connected expect(ftpClient.isConnected()).andReturn(true).anyTimes(); - // first run FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2", "test3"); - expect(ftpClient.listFiles()).andReturn(mockedFTPFiles); + + // expect two receive runs + expect(ftpClient.listFiles()).andReturn(mockedFTPFiles).times(2); + ftpClient.disconnect(); + expectLastCall().times(2); + + // first run expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true); expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true); - - ftpClient.disconnect(); // second run - expect(ftpClient.listFiles()).andReturn(mockedFTPFiles); expect(ftpClient.retrieveFile(eq("test3"), isA(OutputStream.class))).andReturn(true); + // create message expect(messageCreator.createMessage(isA(List.class))).andAnswer(new IAnswer>>() { public Message> answer() throws Throwable { @@ -202,16 +208,71 @@ public class FtpSourceTests { } }).times(2); - ftpClient.disconnect(); - replay(globalMocks); Message> receivedFiles1 = ftpSource.receive(); ftpSource.onSend(receivedFiles1); Message> receivedFiles2 = ftpSource.receive(); ftpSource.onSend(receivedFiles2); verify(globalMocks); + List allReceived = new ArrayList(receivedFiles1.getPayload()); + allReceived.addAll(receivedFiles2.getPayload()); assertEquals(2, receivedFiles1.getPayload().size()); assertEquals(1, receivedFiles2.getPayload().size()); + assertTrue(allReceived.containsAll(Arrays.asList(new File[] { new File("test1"), new File("test2"), + new File("test3") }))); + } + + @Test + public void concurrentPollingSunnyDay() throws Exception { + + this.ftpSource.setMaxMessagesPerPayload(2); + // assume client already connected + expect(ftpClient.isConnected()).andReturn(true).anyTimes(); + // first run + FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2", "test3", "test4", "test5"); + expect(ftpClient.listFiles()).andReturn(mockedFTPFiles); + expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true); + expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true); + + // second poll + expect(ftpClient.listFiles()).andReturn(mockedFTPFiles); + expect(ftpClient.retrieveFile(eq("test3"), isA(OutputStream.class))).andReturn(true); + expect(ftpClient.retrieveFile(eq("test4"), isA(OutputStream.class))).andReturn(true); + + expect(ftpClient.listFiles()).andReturn(mockedFTPFiles); + expect(ftpClient.retrieveFile(eq("test5"), isA(OutputStream.class))).andReturn(true); + // create message + expect(messageCreator.createMessage(isA(List.class))).andAnswer(new IAnswer>>() { + public Message> answer() throws Throwable { + return new GenericMessage(getCurrentArguments()[0]); + } + }).times(3); + + ftpClient.disconnect(); + expectLastCall().times(3); + + replay(globalMocks); + + final CountDownLatch receivesDone = new CountDownLatch(3); + + for (int i = 0; i < 3; i++) { + new Thread(new Runnable() { + public void run() { + Message> recievedFiles = ftpSource.receive(); + receivesDone.countDown(); + try { + receivesDone.await(); + } + catch (InterruptedException e) { + } + finally { + ftpSource.onSend(recievedFiles); + } + } + }).start(); + } + receivesDone.await(); + verify(globalMocks); } @AfterClass