From d6aed9594880cd93e8930911e6539f2af4c811e5 Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Sun, 17 Aug 2008 05:37:03 +0000 Subject: [PATCH] Finishing up INT-293, INT-154. Parametrized DefaultMessageMapper, refactored FtpSource to use a pool, added namespace support for FtpTarget. --- .../.classpath | 1 + .../integration/adapter/ftp/FtpSource.java | 149 ++++-------------- .../integration/adapter/ftp/FtpTarget.java | 12 +- .../adapter/ftp/QueuedFTPClientPool.java | 51 +++++- .../adapter/ftp/config/FtpSourceParser.java | 42 ++++- .../adapter/ftp/FtpSourceTests.java | 87 +++------- .../adapter/ftp/FtpTargetTest.java | 3 +- .../ftp/config/CollectionSplitter.java | 16 ++ .../ftp/config/FtpSourceIntegrationTests.java | 12 +- .../ftp/config/FtpSourceParserTests.java | 36 +++-- .../ftp/config/FtpTargetIntegrationTest.java | 19 +-- .../config/ftpSourceWithChannelAdapter.xml | 20 ++- .../message/DefaultMessageMapper.java | 4 +- 13 files changed, 211 insertions(+), 241 deletions(-) create mode 100644 org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/CollectionSplitter.java diff --git a/org.springframework.integration.adapter/.classpath b/org.springframework.integration.adapter/.classpath index b3fa2c313c..8c16497102 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/ftp/FtpSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java index 4c24f4c827..181cd698ac 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 @@ -30,7 +30,6 @@ import org.apache.commons.logging.LogFactory; import org.apache.commons.net.ftp.FTP; import org.apache.commons.net.ftp.FTPClient; 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.Backlog; @@ -48,56 +47,19 @@ import org.springframework.util.StringUtils; * @author Mark Fisher * @author Iwein Fuld */ -public class FtpSource extends AbstractDirectorySource> implements DisposableBean { - - private final static String DEFAULT_HOST = "localhost"; - - private final static int DEFAULT_PORT = 21; - - private final static String DEFAULT_REMOTE_WORKING_DIRECTORY = "/"; +public class FtpSource extends AbstractDirectorySource> { private final Log logger = LogFactory.getLog(this.getClass()); - private volatile String username; - - private volatile String password; - - private volatile String host = DEFAULT_HOST; - - private volatile int port = DEFAULT_PORT; - - private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY; - private volatile File localWorkingDirectory; private int maxFilesPerPayload = -1; - private final FTPClient client; + private final FTPClientPool clientPool; - public FtpSource(MessageCreator, List> messageCreator) { - this(messageCreator, new FTPClient()); - } - - public FtpSource(MessageCreator, List> messageCreator, FTPClient client) { + public FtpSource(MessageCreator, List> messageCreator, FTPClientPool clientPool) { super(messageCreator); - this.client = client; - } - - public void setHost(String host) { - this.host = host; - } - - public void setPort(int port) { - this.port = port; - } - - @Required - public void setUsername(String username) { - this.username = username; - } - - public void setPassword(String password) { - this.password = password; + this.clientPool = clientPool; } public void setMaxMessagesPerPayload(int maxMessagesPerPayload) { @@ -105,12 +67,6 @@ public class FtpSource extends AbstractDirectorySource> implements Di this.maxFilesPerPayload = maxMessagesPerPayload; } - public void setRemoteWorkingDirectory(String remoteWorkingDirectory) { - Assert.notNull(remoteWorkingDirectory, "'remoteWorkingDirectory' cannot be null"); - // FtpClient is picky about "", so we make it happy - this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/"); - } - public void setLocalWorkingDirectory(File localWorkingDirectory) { Assert.notNull(localWorkingDirectory, "'localWorkingDirectory' must not be null"); this.localWorkingDirectory = localWorkingDirectory; @@ -130,87 +86,48 @@ public class FtpSource extends AbstractDirectorySource> implements Di @Override protected void populateSnapshot(Map snapshot) throws IOException { - establishConnection(); - FTPFile[] fileList = this.client.listFiles(); - - for (FTPFile ftpFile : fileList) { - /* - * 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); + FTPClient client = clientPool.getClient(); + FTPFile[] fileList = client.listFiles(); + try { + for (FTPFile ftpFile : fileList) { + /* + * 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); + } } } - } - - protected void establishConnection() throws IOException { - if (this.client.isConnected()) { - if (logger.isDebugEnabled()) { - logger.debug("client already connected"); - } - return; - } - if (!StringUtils.hasText(this.username)) { - throw new MessagingException("username is required"); - } - this.client.connect(this.host, this.port); - if (!this.client.login(this.username, this.password)) { - throw new MessagingException("Login failed. Please check the username and password."); - } - if (logger.isDebugEnabled()) { - logger.debug("login successful"); - } - this.client.setFileType(FTP.IMAGE_FILE_TYPE); - if (!this.remoteWorkingDirectory.equals(this.client.printWorkingDirectory()) - && !this.client.changeWorkingDirectory(this.remoteWorkingDirectory)) { - throw new MessagingException("Could not change directory to '" + remoteWorkingDirectory - + "'. Please check the path."); - } - if (logger.isDebugEnabled()) { - logger.debug("working directory is: " + this.client.printWorkingDirectory()); + finally { + clientPool.releaseClient(client); } } protected List retrieveNextPayload() throws IOException { - establishConnection(); - List files = new ArrayList(); - Set toDo = this.getDirectoryContentManager().getProcessingBuffer().keySet(); - for (String fileName : toDo) { - File file = new File(this.localWorkingDirectory, fileName); - if (file.exists()) { - file.delete(); - } - FileOutputStream fileOutputStream = new FileOutputStream(file); - this.client.retrieveFile(fileName, fileOutputStream); - fileOutputStream.close(); - files.add(file); - } - disconnect(); - return files; - } - - protected void disconnect() { + FTPClient client = clientPool.getClient(); try { - if (this.client.isConnected()) { - this.client.disconnect(); - if (logger.isDebugEnabled()) { - logger.debug("connection closed"); + List files = new ArrayList(); + Set toDo = this.getDirectoryContentManager().getProcessingBuffer().keySet(); + for (String fileName : toDo) { + File file = new File(this.localWorkingDirectory, fileName); + if (file.exists()) { + file.delete(); } + FileOutputStream fileOutputStream = new FileOutputStream(file); + client.retrieveFile(fileName, fileOutputStream); + fileOutputStream.close(); + files.add(file); } + return files; } - catch (IOException ioe) { - if (logger.isErrorEnabled()) { - logger.error("Error when disconnecting from ftp.", ioe); - } + finally { + clientPool.releaseClient(client); } } - public void destroy() throws Exception { - disconnect(); - } @SuppressWarnings("unchecked") @Override diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpTarget.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpTarget.java index 726ce7b01b..7d27b52559 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpTarget.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpTarget.java @@ -38,18 +38,14 @@ public class FtpTarget implements MessageTarget { private final MessageMapper messageMapper; - private volatile FTPClientPool ftpClientPool = new QueuedFTPClientPool(); + private final FTPClientPool ftpClientPool; - public FtpTarget(MessageMapper messageMapper) { - Assert.notNull(messageMapper, "MessageMapper must not be null"); - this.messageMapper = messageMapper; - } - - - public void setFtpClientPool(FTPClientPool ftpClientPool) { + public FtpTarget(MessageMapper messageMapper, FTPClientPool ftpClientPool) { + Assert.notNull(messageMapper, "messageMapper must not be null"); Assert.notNull(ftpClientPool, "ftpClientPool must not be null"); this.ftpClientPool = ftpClientPool; + this.messageMapper = messageMapper; } public boolean send(Message message) { diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPool.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPool.java index 0482618d50..0689b9a069 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPool.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPool.java @@ -26,8 +26,11 @@ import org.apache.commons.logging.LogFactory; import org.apache.commons.net.ftp.FTP; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPClientConfig; +import org.apache.commons.net.ftp.FTPReply; +import org.springframework.integration.message.MessagingException; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * FTPClientPool implementation based on a Queue. This implementation has a @@ -39,6 +42,8 @@ public class QueuedFTPClientPool implements FTPClientPool { private static final int DEFAULT_POOL_SIZE = 5; + private static final String DEFAULT_REMOTE_WORKING_DIRECTORY = "/"; + private final Queue pool; private volatile FTPClientConfig config; @@ -47,14 +52,16 @@ public class QueuedFTPClientPool implements FTPClientPool { private volatile int port = FTP.DEFAULT_PORT; - private volatile String user; + private volatile String username; - private volatile String pass; + private volatile String password; private volatile FTPClientFactory factory = new DefaultFactory(); private final Log log = LogFactory.getLog(this.getClass()); + private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY; + public QueuedFTPClientPool() { this(DEFAULT_POOL_SIZE); } @@ -98,14 +105,19 @@ public class QueuedFTPClientPool implements FTPClientPool { this.port = port; } - public void setUser(String user) { + public void setUsername(String user) { Assert.hasText(user); - this.user = user; + this.username = user; } - public void setPass(String pass) { + public void setPassword(String pass) { Assert.notNull(pass); - this.pass = pass; + this.password = pass; + } + + public void setRemoteWorkingDirectory(String remoteWorkingDirectory) { + Assert.notNull(remoteWorkingDirectory); + this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/"); } public void setFactory(FTPClientFactory factory) { @@ -118,8 +130,33 @@ public class QueuedFTPClientPool implements FTPClientPool { public FTPClient getClient() throws SocketException, IOException { FTPClient client = new FTPClient(); client.configure(config); + if (!StringUtils.hasText(username)) { + throw new MessagingException("username is required"); + } client.connect(host, port); - client.login(user, pass); + if (!FTPReply.isPositiveCompletion(client.getReplyCode())) { + throw new MessagingException("Connecting to server [" + host + ":" + port + + "] failed, please check the connection"); + } + if (log.isDebugEnabled()) { + log.debug("Connected to server [" + host + ":" + port + "]"); + } + if (!client.login(username, password)) { + throw new MessagingException("Login failed. Please check the username and password."); + } + if (log.isDebugEnabled()) { + log.debug("login successful"); + } + client.setFileType(FTP.BINARY_FILE_TYPE); + + if (!remoteWorkingDirectory.equals(client.printWorkingDirectory()) + && !client.changeWorkingDirectory(remoteWorkingDirectory)) { + throw new MessagingException("Could not change directory to '" + remoteWorkingDirectory + + "'. Please check the path."); + } + if (log.isDebugEnabled()) { + log.debug("working directory is: " + client.printWorkingDirectory()); + } return client; } } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java index 980c770469..5c16e10867 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java @@ -16,27 +16,63 @@ package org.springframework.integration.adapter.ftp.config; -import org.w3c.dom.Element; - +import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.integration.adapter.file.config.AbstractDirectorySourceParser; import org.springframework.integration.adapter.ftp.FtpSource; +import org.springframework.integration.adapter.ftp.QueuedFTPClientPool; +import org.w3c.dom.Element; /** * Parser for the <ftp-source/> element. * * @author Mark Fisher * @author Marius Bogoevici + * @author Iwein Fuld */ public class FtpSourceParser extends AbstractDirectorySourceParser { + private static final String POOL_ATTRIBUTE_USER = "username"; + + private static final String POOL_ATTRIBUTE_PASS = "password"; + + private static final String POOL_ATTRIBUTE_HOST = "host"; + + private static final String POOL_ATTRIBUTE_PORT = "port"; + + private static final String POOL_ATTRIBUTE_REMOTEDIR = "remote-working-directory"; public FtpSourceParser() { super(true); } - @Override protected Class getBeanClass(Element element) { return FtpSource.class; } + @Override + protected boolean isEligibleAttribute(String attributeName) { + return !POOL_ATTRIBUTE_HOST.equals(attributeName) + && !POOL_ATTRIBUTE_PASS.equals(attributeName) + && !POOL_ATTRIBUTE_PORT.equals(attributeName) + && !POOL_ATTRIBUTE_USER.equals(attributeName) + && !POOL_ATTRIBUTE_REMOTEDIR.equals(attributeName) + && super.isEligibleAttribute(attributeName); + } + + @Override + protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) { + super.postProcess(beanDefinition, element); + String user = element.getAttribute(POOL_ATTRIBUTE_USER); + String pass = element.getAttribute(POOL_ATTRIBUTE_PASS); + String host = element.getAttribute(POOL_ATTRIBUTE_HOST); + String port = element.getAttribute(POOL_ATTRIBUTE_PORT); + String remoteWorkingDirectory = element.getAttribute(POOL_ATTRIBUTE_REMOTEDIR); + QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool(); + queuedFTPClientPool.setUsername(user); + queuedFTPClientPool.setPassword(pass); + queuedFTPClientPool.setHost(host); + queuedFTPClientPool.setPort(Integer.parseInt(port)); + queuedFTPClientPool.setRemoteWorkingDirectory(remoteWorkingDirectory); + beanDefinition.addConstructorArgValue(queuedFTPClientPool); + } } 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 976b344831..4c5355d688 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,18 +16,10 @@ package org.springframework.integration.adapter.ftp; -import static org.easymock.EasyMock.*; -import static org.easymock.EasyMock.eq; -import static org.easymock.EasyMock.expect; -import static org.easymock.EasyMock.getCurrentArguments; -import static org.easymock.EasyMock.isA; -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.easymock.classextension.EasyMock.*; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import java.io.File; import java.io.FilenameFilter; @@ -59,7 +51,14 @@ public class FtpSourceTests { private FTPFile ftpFile = createMock(FTPFile.class); - private Object[] globalMocks = new Object[] { messageCreator, ftpClient, ftpFile }; + private FTPClientPool ftpClientPool = createNiceMock(FTPClientPool.class); + + @Before + public void liberalPool() throws Exception { + expect(ftpClientPool.getClient()).andReturn(ftpClient).anyTimes(); + } + + private Object[] globalMocks = new Object[] { messageCreator, ftpClient, ftpFile, ftpClientPool }; private static final String HOST = "testHost"; @@ -73,10 +72,7 @@ public class FtpSourceTests { @Before public void initializeFtpSource() { - ftpSource = new FtpSource(messageCreator, ftpClient); - ftpSource.setHost(HOST); - ftpSource.setUsername(USER); - ftpSource.setPassword(PASS); + ftpSource = new FtpSource(messageCreator, ftpClientPool); } @Before @@ -86,20 +82,12 @@ public class FtpSourceTests { @Test public void retrieveSingleFile() throws Exception { - // connect client and get file - expect(ftpClient.isConnected()).andReturn(false); - expect(ftpClient.isConnected()).andReturn(true).anyTimes(); - ftpClient.connect(HOST, 21); - expect(ftpClient.login(USER, PASS)).andReturn(true); - expect(ftpClient.setFileType(anyInt())).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")))); - ftpClient.disconnect(); replay(globalMocks); Message> received = ftpSource.receive(); ftpSource.onSend(received); @@ -124,15 +112,6 @@ public class FtpSourceTests { @Test public void retrieveMultipleFiles() throws Exception { - // connect client - expect(ftpClient.isConnected()).andReturn(false); - expect(ftpClient.isConnected()).andReturn(true).anyTimes(); - - ftpClient.connect(HOST, 21); - expect(ftpClient.login(USER, PASS)).andReturn(true); - expect(ftpClient.setFileType(anyInt())).andReturn(true); - expect(ftpClient.printWorkingDirectory()).andReturn("/").anyTimes(); - // get files expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test1", "test2")).times(2); @@ -141,7 +120,6 @@ public class FtpSourceTests { // create message List files = Arrays.asList(new File("test1"), new File("test2")); expect(messageCreator.createMessage(isA(List.class))).andReturn(new GenericMessage(files)); - ftpClient.disconnect(); replay(globalMocks); Message receivedFiles = ftpSource.receive(); @@ -154,16 +132,12 @@ public class FtpSourceTests { @Test public void retrieveMultipleChangingFiles() throws Exception { - - // assume client already connected - expect(ftpClient.isConnected()).andReturn(true).anyTimes(); // first run FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2"); 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); - ftpClient.disconnect(); // second run, change the date so the messages should be retrieved again // expect(ftpClient.isConnected()).andReturn(true); FTPFile[] mockedFTPFiles2 = mockedFTPFilesNamed("test1", "test2"); @@ -173,7 +147,6 @@ public class FtpSourceTests { // create message List files = Arrays.asList(new File("test1"), new File("test2")); expect(messageCreator.createMessage(isA(List.class))).andReturn(new GenericMessage(files)).times(2); - ftpClient.disconnect(); replay(globalMocks); Message receivedFiles = ftpSource.receive(); @@ -188,13 +161,10 @@ public class FtpSourceTests { this.ftpSource.setMaxMessagesPerPayload(2); // assume client already connected - expect(ftpClient.isConnected()).andReturn(true).anyTimes(); FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2", "test3"); // 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); @@ -227,8 +197,6 @@ public class FtpSourceTests { 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); @@ -249,9 +217,6 @@ public class FtpSourceTests { } }).times(3); - ftpClient.disconnect(); - expectLastCall().times(3); - replay(globalMocks); final CountDownLatch receivesDone = new CountDownLatch(3); @@ -275,24 +240,20 @@ 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); - } + @Test + public void onFailure() throws Exception { + 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); + 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/FtpTargetTest.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpTargetTest.java index ad19349382..3dff09942f 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpTargetTest.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpTargetTest.java @@ -64,8 +64,7 @@ public class FtpTargetTest { @Before public void intitializeSubject() { - this.ftpTarget = new FtpTarget(messageMapper); - ftpTarget.setFtpClientPool(ftpClientPool); + this.ftpTarget = new FtpTarget(messageMapper, ftpClientPool); } // Tests diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/CollectionSplitter.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/CollectionSplitter.java new file mode 100644 index 0000000000..f93efef741 --- /dev/null +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/CollectionSplitter.java @@ -0,0 +1,16 @@ +package org.springframework.integration.adapter.ftp.config; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.springframework.integration.annotation.Splitter; +import org.springframework.integration.message.Message; + +public class CollectionSplitter { + + @SuppressWarnings("unchecked") + @Splitter public List split(Message message) { + return new ArrayList(message.getPayload()); + } +} 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 c7e3810c55..23f94a83ed 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 @@ -14,6 +14,7 @@ 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.adapter.ftp.QueuedFTPClientPool; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.config.MessageBusParser; @@ -57,12 +58,13 @@ public class FtpSourceIntegrationTests { @Before public void initializeFtpSource() throws Exception { - ftpSource = new FtpSource(messageCreator); - ftpSource.setHost("localhost"); - ftpSource.setUsername("ftp-user"); - ftpSource.setPassword("kaas"); + QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool(); + ftpSource = new FtpSource(messageCreator, queuedFTPClientPool); + queuedFTPClientPool.setHost("localhost"); + queuedFTPClientPool.setUsername("ftp-user"); + queuedFTPClientPool.setPassword("kaas"); ftpSource.setLocalWorkingDirectory(localWorkDir); - ftpSource.setRemoteWorkingDirectory("ftp-test"); + queuedFTPClientPool.setRemoteWorkingDirectory("ftp-test"); } @SuppressWarnings("unchecked") diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java index e575d0740d..6d9096fa36 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java @@ -38,6 +38,7 @@ import org.springframework.integration.adapter.ftp.FtpSource; /** * @author Mark Fisher * @author Marius Bogoevici + * @author Iwein Fuld */ public class FtpSourceParserTests { @@ -45,14 +46,15 @@ public class FtpSourceParserTests { public void testFtpSourceAdapterParser() { ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceDefault"); - DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor accessor = new DirectFieldAccessor(sourceAccessor.getPropertyValue("clientPool")); assertEquals("testHost", accessor.getPropertyValue("host")); assertEquals(2121, accessor.getPropertyValue("port")); - assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals(new File("/local"), sourceAccessor.getPropertyValue("localWorkingDirectory")); assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); assertEquals("testUser", accessor.getPropertyValue("username")); assertEquals("testPassword", accessor.getPropertyValue("password")); - Object messageCreator = accessor.getPropertyValue("messageCreator"); + Object messageCreator = sourceAccessor.getPropertyValue("messageCreator"); assertTrue(messageCreator instanceof FileMessageCreator); DirectFieldAccessor messageCreatorAccessor = new DirectFieldAccessor(messageCreator); assertEquals(messageCreatorAccessor.getPropertyValue("deleteFileAfterCreation"), false); @@ -62,14 +64,15 @@ public class FtpSourceParserTests { public void testFtpSourceTextType() { ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceText"); - DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor accessor = new DirectFieldAccessor(sourceAccessor.getPropertyValue("clientPool")); assertEquals("testHost", accessor.getPropertyValue("host")); assertEquals(2121, accessor.getPropertyValue("port")); - assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals(new File("/local"), sourceAccessor.getPropertyValue("localWorkingDirectory")); assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); assertEquals("testUser", accessor.getPropertyValue("username")); assertEquals("testPassword", accessor.getPropertyValue("password")); - Object messageCreator = accessor.getPropertyValue("messageCreator"); + Object messageCreator = sourceAccessor.getPropertyValue("messageCreator"); assertTrue(messageCreator instanceof TextFileMessageCreator); DirectFieldAccessor messageCreatorAccessor = new DirectFieldAccessor(messageCreator); assertEquals(messageCreatorAccessor.getPropertyValue("deleteFileAfterCreation"), true); @@ -79,14 +82,15 @@ public class FtpSourceParserTests { public void testFtpSourceBinaryType() { ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceBinary"); - DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor accessor = new DirectFieldAccessor(sourceAccessor.getPropertyValue("clientPool")); assertEquals("testHost", accessor.getPropertyValue("host")); assertEquals(2121, accessor.getPropertyValue("port")); - assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals(new File("/local"), sourceAccessor.getPropertyValue("localWorkingDirectory")); assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); assertEquals("testUser", accessor.getPropertyValue("username")); assertEquals("testPassword", accessor.getPropertyValue("password")); - Object messageCreator = accessor.getPropertyValue("messageCreator"); + Object messageCreator = sourceAccessor.getPropertyValue("messageCreator"); assertTrue(messageCreator instanceof ByteArrayFileMessageCreator); DirectFieldAccessor messageCreatorAccessor = new DirectFieldAccessor(messageCreator); assertEquals(messageCreatorAccessor.getPropertyValue("deleteFileAfterCreation"), true); @@ -96,14 +100,15 @@ public class FtpSourceParserTests { public void testFtpSourceFileType() { ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceFile"); - DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor accessor = new DirectFieldAccessor(sourceAccessor.getPropertyValue("clientPool")); assertEquals("testHost", accessor.getPropertyValue("host")); assertEquals(2121, accessor.getPropertyValue("port")); - assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals(new File("/local"), sourceAccessor.getPropertyValue("localWorkingDirectory")); assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); assertEquals("testUser", accessor.getPropertyValue("username")); assertEquals("testPassword", accessor.getPropertyValue("password")); - Object messageCreator = accessor.getPropertyValue("messageCreator"); + Object messageCreator = sourceAccessor.getPropertyValue("messageCreator"); assertTrue(messageCreator instanceof FileMessageCreator); DirectFieldAccessor messageCreatorAccessor = new DirectFieldAccessor(messageCreator); assertEquals(messageCreatorAccessor.getPropertyValue("deleteFileAfterCreation"), false); @@ -113,14 +118,15 @@ public class FtpSourceParserTests { public void testFtpSourceCustomType() { ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceCustom"); - DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(ftpSource); + DirectFieldAccessor accessor = new DirectFieldAccessor(sourceAccessor.getPropertyValue("clientPool")); assertEquals("testHost", accessor.getPropertyValue("host")); assertEquals(2121, accessor.getPropertyValue("port")); - assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals(new File("/local"), sourceAccessor.getPropertyValue("localWorkingDirectory")); assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); assertEquals("testUser", accessor.getPropertyValue("username")); assertEquals("testPassword", accessor.getPropertyValue("password")); - Object messageCreator = accessor.getPropertyValue("messageCreator"); + Object messageCreator = sourceAccessor.getPropertyValue("messageCreator"); assertTrue(messageCreator instanceof CustomMessageCreator); // not testing for deleteFileAfterCreation - this is completely left up // to the implementation diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpTargetIntegrationTest.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpTargetIntegrationTest.java index f00fe016c4..a0925678a7 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpTargetIntegrationTest.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpTargetIntegrationTest.java @@ -25,32 +25,27 @@ import org.junit.AfterClass; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; -import org.springframework.integration.adapter.ftp.QueuedFTPClientPool; import org.springframework.integration.adapter.ftp.FtpTarget; +import org.springframework.integration.adapter.ftp.QueuedFTPClientPool; +import org.springframework.integration.message.DefaultMessageMapper; import org.springframework.integration.message.GenericMessage; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageMapper; /** * @author Iwein Fuld */ -@Ignore +//@Ignore public class FtpTargetIntegrationTest { private FtpTarget ftpTarget; @Before public void initFtpTarget() { - ftpTarget = new FtpTarget(new MessageMapper() { - public File mapMessage(Message message) { - return message.getPayload(); - } - }); QueuedFTPClientPool clientPool = new QueuedFTPClientPool(); clientPool.setHost("localhost"); - clientPool.setUser("ftp-user"); - clientPool.setPass("kaas"); - ftpTarget.setFtpClientPool(clientPool); + clientPool.setUsername("ftp-user"); + clientPool.setPassword("kaas"); + clientPool.setRemoteWorkingDirectory("ftp-test"); + ftpTarget = new FtpTarget(new DefaultMessageMapper(), clientPool); } @Test 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 index 0b267a5c4e..8c37b148c5 100644 --- 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 @@ -12,21 +12,25 @@ + class="org.springframework.integration.adapter.ftp.FtpSource"> + + + - + + - - - + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/DefaultMessageMapper.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/DefaultMessageMapper.java index 2b4fe8afee..7c8c96d0e1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/DefaultMessageMapper.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/DefaultMessageMapper.java @@ -22,9 +22,9 @@ package org.springframework.integration.message; * * @author Mark Fisher */ -public class DefaultMessageMapper implements MessageMapper { +public class DefaultMessageMapper implements MessageMapper { - public Object mapMessage(Message message) { + public T mapMessage(Message message) { return (message != null) ? message.getPayload() : null; }