diff --git a/org.springframework.integration.adapter/.classpath b/org.springframework.integration.adapter/.classpath index 66fbd841a2..e2e0bb92c8 100644 --- a/org.springframework.integration.adapter/.classpath +++ b/org.springframework.integration.adapter/.classpath @@ -26,5 +26,6 @@ + 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 5f516e0624..df41747ffe 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 @@ -110,6 +110,7 @@ public abstract class AbstractDirectorySource implements PollableSource, M if (this.logger.isWarnEnabled()) { logger.warn("Failure notification received by " + this.getClass().getSimpleName(), exception); } + directoryContentManager.processingFailed(); } /** @@ -126,8 +127,8 @@ public abstract class AbstractDirectorySource implements PollableSource, M */ protected abstract T retrieveNextPayload() throws IOException; - protected final void fileProcessed(String fileName) { - this.directoryContentManager.fileProcessed(fileName); + protected final void fileProcessed(String ... fileNames) { + this.directoryContentManager.fileProcessed(fileNames); } } 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 457f9cbdb9..ad68cf5fc0 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 @@ -61,7 +61,7 @@ public class DirectoryContentManager { * messages in the processing buffer something is wrong because they * were not processed, nor raised as failed. */ - Assert.isTrue(processingBuffer.get().isEmpty()); + Assert.isTrue(processingBuffer.get().isEmpty(), "Processing buffer not emptied before poll."); Iterator> iter = this.backlog.entrySet().iterator(); while (iter.hasNext()) { String fileName = iter.next().getKey(); @@ -97,15 +97,13 @@ public class DirectoryContentManager { } } - 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 processingFailed() { + if (logger.isDebugEnabled()) { + logger.debug("Moving all files from processing buffer to backlog. Processing has failed"); } + Map processing = this.processingBuffer.get(); + this.backlog.putAll(processing); + processing.clear(); } public synchronized void fileProcessed(String... fileNames) { 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 37fb864e76..976b344831 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 @@ -48,6 +48,7 @@ import org.junit.Test; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageCreator; +import org.springframework.integration.message.MessagingException; @SuppressWarnings("unchecked") public class FtpSourceTests { @@ -92,9 +93,9 @@ public class FtpSourceTests { ftpClient.connect(HOST, 21); expect(ftpClient.login(USER, PASS)).andReturn(true); expect(ftpClient.setFileType(anyInt())).andReturn(true); - expect(ftpClient.printWorkingDirectory()).andReturn("/"); - expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test")); - expect(ftpClient.retrieveFile(eq("test"), isA(OutputStream.class))).andReturn(true); + expect(ftpClient.printWorkingDirectory()).andReturn("/").anyTimes(); + expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test1")); + expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true); // create message expect(messageCreator.createMessage(isA(List.class))).andReturn( new GenericMessage(Arrays.asList(new File("test1")))); @@ -130,7 +131,7 @@ public class FtpSourceTests { ftpClient.connect(HOST, 21); expect(ftpClient.login(USER, PASS)).andReturn(true); expect(ftpClient.setFileType(anyInt())).andReturn(true); - expect(ftpClient.printWorkingDirectory()).andReturn("/"); + expect(ftpClient.printWorkingDirectory()).andReturn("/").anyTimes(); // get files expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test1", "test2")).times(2); @@ -274,6 +275,24 @@ public class FtpSourceTests { receivesDone.await(); verify(globalMocks); } + + @Test public void onFailure() throws Exception{ + // connect client and get file + expect(ftpClient.isConnected()).andReturn(true).anyTimes(); + + expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test1")).times(2); + expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true).times(2); + // create message + expect(messageCreator.createMessage(isA(List.class))).andReturn( + new GenericMessage(Arrays.asList(new File("test1")))).times(2); + ftpClient.disconnect(); + ftpClient.disconnect(); + replay(globalMocks); + Message> received = ftpSource.receive(); + ftpSource.onFailure(new MessagingException(received)); + assertEquals(received, ftpSource.receive()); + verify(globalMocks); + } @AfterClass public static void deleteFiles() { 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 5cf6a09ff7..c7e3810c55 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 @@ -3,13 +3,20 @@ package org.springframework.integration.adapter.ftp.config; import static org.junit.Assert.assertTrue; import java.io.File; +import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import org.junit.Before; import org.junit.BeforeClass; import org.junit.Ignore; import org.junit.Test; +import org.springframework.context.ApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.adapter.ftp.FtpSource; +import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.config.MessageBusParser; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageCreator; @@ -64,6 +71,15 @@ public class FtpSourceIntegrationTests { Message> received = ftpSource.receive(); assertTrue(received.getPayload().iterator().next().exists()); } - + @Test public void withChannelAdapter() { + ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceWithChannelAdapter.xml", this.getClass()); + ChannelRegistry channelRegistry = (ChannelRegistry) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); + PollableChannel input = (PollableChannel) channelRegistry.lookupChannel("output"); + List files = new ArrayList(); + files.add((File) input.receive().getPayload()); + files.add((File) input.receive().getPayload()); + assertTrue(files.containsAll(Arrays.asList(new File("file1"), new File("file2")))); + } + } diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceWithChannelAdapter.xml b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceWithChannelAdapter.xml new file mode 100644 index 0000000000..0b267a5c4e --- /dev/null +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceWithChannelAdapter.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/util/CollectionSplitter.java b/org.springframework.integration/src/main/java/org/springframework/integration/util/CollectionSplitter.java new file mode 100644 index 0000000000..f2a1c2aa2d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/util/CollectionSplitter.java @@ -0,0 +1,22 @@ +package org.springframework.integration.util; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.springframework.integration.annotation.Splitter; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; + +public class CollectionSplitter { + + @SuppressWarnings("unchecked") + @Splitter + public final List split(Message message) { + List splitMessages = new ArrayList(); + for (Object payload : ((Collection) message.getPayload())) { + splitMessages.add(new GenericMessage(payload)); + } + return splitMessages; + } +}