diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java index 8513f7857a..393c24467f 100644 --- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java +++ b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java @@ -25,11 +25,10 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageCreator; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.PollableSource; -import org.springframework.util.Assert; /** * Base class for implementing a PollableSource that creates messages from files @@ -46,17 +45,13 @@ public abstract class AbstractDirectorySource implements PollableSource, M private final Backlog backlog; - private final MessageCreator messageCreator; - - public AbstractDirectorySource(MessageCreator messageCreator) { - this(messageCreator, null); + public AbstractDirectorySource() { + this(null); } - public AbstractDirectorySource(MessageCreator messageCreator, Comparator comparator) { + public AbstractDirectorySource(Comparator comparator) { this.backlog = comparator == null ? new Backlog() : new Backlog(comparator); - Assert.notNull(messageCreator, "The MessageCreator must not be null"); - this.messageCreator = messageCreator; } @@ -64,10 +59,6 @@ public abstract class AbstractDirectorySource implements PollableSource, M return this.backlog; } - public MessageCreator getMessageCreator() { - return this.messageCreator; - } - public final Message receive() { try { refreshSnapshotAndMarkProcessing(this.backlog); @@ -89,17 +80,16 @@ public abstract class AbstractDirectorySource implements PollableSource, M /** * Hook point for implementors to create the next message that should be - * received. Implementations can use a File by File approach (like - * FileSource). In cases where retrieval could be expensive because of - * network latency, a batched approach could be implemented here. See - * FtpSource for an example. + * received. Implementations can use a File by File approach or in cases + * where retrieval could be expensive because of network latency, a batched + * approach could be implemented here. See FtpSource for an example. * * @return the next message containing (part of) the unprocessed content of * the directory * @throws IOException */ protected Message buildNextMessage() throws IOException { - return this.messageCreator.createMessage(retrieveNextPayload()); + return MessageBuilder.withPayload(retrieveNextPayload()).build(); } public void onSend(Message message) { diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java index 2313568b0c..39f97985bd 100644 --- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java +++ b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java @@ -25,8 +25,6 @@ import java.util.List; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; -import org.springframework.integration.message.DefaultMessageCreator; -import org.springframework.integration.message.MessageCreator; import org.springframework.util.Assert; /** @@ -46,11 +44,6 @@ public class FtpSource extends AbstractDirectorySource> { public FtpSource(FTPClientPool clientPool) { - this(new DefaultMessageCreator>(), clientPool); - } - - public FtpSource(MessageCreator, List> messageCreator, FTPClientPool clientPool) { - super(messageCreator); this.clientPool = clientPool; } @@ -84,8 +77,8 @@ public class FtpSource extends AbstractDirectorySource> { * if files couldn't be parsed */ if (ftpFile != null) { - FileSnapshot fileSnapshot = new FileSnapshot(ftpFile.getName(), ftpFile.getTimestamp() - .getTimeInMillis(), ftpFile.getSize()); + FileSnapshot fileSnapshot = new FileSnapshot(ftpFile.getName(), + ftpFile.getTimestamp().getTimeInMillis(), ftpFile.getSize()); snapshot.add(fileSnapshot); } } @@ -101,8 +94,7 @@ public class FtpSource extends AbstractDirectorySource> { List files = new ArrayList(); List toDo = this.getBacklog().getProcessingBuffer(); for (FileSnapshot fileSnapshot : toDo) { - // some awkwardness here because the local path may be different - // from the remote path + // local path may be different from the remote path File file = new File(this.localWorkingDirectory, fileSnapshot.getFileName()); if (file.exists()) { file.delete(); diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java index 03a2b1a04b..c9f9e90714 100644 --- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java +++ b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java @@ -47,6 +47,7 @@ public class QueuedFTPClientPool implements FTPClientPool { private static final String DEFAULT_REMOTE_WORKING_DIRECTORY = "/"; + private final Queue pool; private volatile FTPClientConfig config; @@ -59,13 +60,25 @@ public class QueuedFTPClientPool implements FTPClientPool { private volatile String password; - private volatile FTPClientFactory factory = new DefaultFactory(); + private volatile FTPClientFactory factory = new DefaultFTPClientFactory(); private final Log log = LogFactory.getLog(this.getClass()); private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY; - // setters + + public QueuedFTPClientPool() { + this(DEFAULT_POOL_SIZE); + } + + /** + * @param maxPoolSize the maximum size of the pool + */ + public QueuedFTPClientPool(int maxPoolSize) { + pool = new ArrayBlockingQueue(maxPoolSize); + } + + public void setConfig(FTPClientConfig config) { Assert.notNull(config); this.config = config; @@ -101,22 +114,11 @@ public class QueuedFTPClientPool implements FTPClientPool { this.factory = factory; } - public QueuedFTPClientPool() { - this(DEFAULT_POOL_SIZE); - } - - /** - * @param maxPoolSize the maximum size of the pool - */ - public QueuedFTPClientPool(int maxPoolSize) { - pool = new ArrayBlockingQueue(maxPoolSize); - } - /** * Returns an active FTPClient connected to the configured server. When no * clients are available in the queue a new client is created with the * factory. - * + *

* It is possible that released clients are disconnected by the remote * server (@see {@link FTPClient#sendNoOp()}. In this case getClient is * called recursively to obtain a client that is still alive. For this @@ -157,7 +159,7 @@ public class QueuedFTPClientPool implements FTPClientPool { } } - private class DefaultFactory implements FTPClientFactory { + private class DefaultFTPClientFactory implements FTPClientFactory { public FTPClient getClient() throws SocketException, IOException { FTPClient client = new FTPClient(); @@ -192,4 +194,5 @@ public class QueuedFTPClientPool implements FTPClientPool { return client; } } + } diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java index 4110acc34a..e4a85b3ac5 100644 --- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java +++ b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java @@ -25,7 +25,6 @@ import org.springframework.integration.config.AbstractPollingInboundChannelAdapt import org.springframework.integration.config.IntegrationNamespaceUtils; import org.springframework.integration.ftp.FtpSource; import org.springframework.integration.ftp.QueuedFTPClientPool; -import org.springframework.util.StringUtils; /** * Parser for the <inbound-channel-adapter/> element of the 'ftp' namespace. @@ -39,11 +38,6 @@ public class FtpInboundChannelAdapterParser extends AbstractPollingInboundChanne @Override protected String parseSource(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(FtpSource.class); - String messageCreatorReference = element.getAttribute("message-creator"); - if (StringUtils.hasText(messageCreatorReference)) { - builder.addConstructorArgReference(messageCreatorReference); - } - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "local-working-directory"); String username = element.getAttribute("username"); String password = element.getAttribute("password"); String host = element.getAttribute("host"); @@ -56,6 +50,7 @@ public class FtpInboundChannelAdapterParser extends AbstractPollingInboundChanne queuedFTPClientPool.setPort(Integer.parseInt(port)); queuedFTPClientPool.setRemoteWorkingDirectory(remoteWorkingDirectory); builder.addConstructorArgValue(queuedFTPClientPool); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "local-working-directory"); return BeanDefinitionReaderUtils.registerWithGeneratedName( builder.getBeanDefinition(), parserContext.getRegistry()); } diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd index b0ecddf4ec..1098325c3f 100644 --- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd +++ b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd @@ -24,13 +24,13 @@ + - diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSourceTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSourceTests.java index e87a4e5b4a..241e1ee03b 100644 --- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSourceTests.java +++ b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSourceTests.java @@ -18,7 +18,6 @@ package org.springframework.integration.ftp; 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.createNiceMock; @@ -41,17 +40,12 @@ import java.util.concurrent.CountDownLatch; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; import org.apache.oro.io.Perl5FilenameFilter; -import org.easymock.IAnswer; import org.junit.AfterClass; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; -import org.springframework.integration.ftp.FTPClientPool; -import org.springframework.integration.ftp.FtpSource; -import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageCreator; /** * @author Iwein Fuld @@ -59,28 +53,27 @@ import org.springframework.integration.message.MessageCreator; @SuppressWarnings("unchecked") public class FtpSourceTests { - private MessageCreator, List> messageCreator = createMock(MessageCreator.class); - private FTPClient ftpClient = createMock(FTPClient.class); private FTPFile ftpFile = createMock(FTPFile.class); 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 Object[] globalMocks = new Object[] { ftpClient, ftpFile, ftpClientPool }; private FtpSource ftpSource; private Long size = 100l; + + @Before + public void liberalPool() throws Exception { + expect(ftpClientPool.getClient()).andReturn(ftpClient).anyTimes(); + } + @Before public void initializeFtpSource() { - ftpSource = new FtpSource(messageCreator, ftpClientPool); + ftpSource = new FtpSource(ftpClientPool); } @Before @@ -88,14 +81,11 @@ public class FtpSourceTests { reset(globalMocks); } + @Test public void retrieveSingleFile() throws Exception { - 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")))); replay(globalMocks); Message> received = ftpSource.receive(); ftpSource.onSend(received); @@ -122,12 +112,9 @@ public class FtpSourceTests { public void retrieveMultipleFiles() throws Exception { // get files expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test1", "test2")).times(2); - expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true); expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true); - // create message List files = Arrays.asList(new File("test1"), new File("test2")); - expect(messageCreator.createMessage(isA(List.class))).andReturn(new GenericMessage(files)); replay(globalMocks); Message receivedFiles = ftpSource.receive(); @@ -147,14 +134,11 @@ public class FtpSourceTests { expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true); // second run, change the date so the messages should be retrieved again - // expect(ftpClient.isConnected()).andReturn(true); FTPFile[] mockedFTPFiles2 = mockedFTPFilesNamed("test1", "test2"); expect(ftpClient.listFiles()).andReturn(mockedFTPFiles2); expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true); expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true); - // create message List files = Arrays.asList(new File("test1"), new File("test2")); - expect(messageCreator.createMessage(isA(List.class))).andReturn(new GenericMessage(files)).times(2); replay(globalMocks); Message receivedFiles = ftpSource.receive(); @@ -180,13 +164,6 @@ public class FtpSourceTests { // second run 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 { - return new GenericMessage(getCurrentArguments()[0]); - } - }).times(2); - replay(globalMocks); Message> receivedFiles1 = ftpSource.receive(); ftpSource.onSend(receivedFiles1); @@ -219,12 +196,6 @@ public class FtpSourceTests { 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); replay(globalMocks); recorded.countDown(); @@ -263,13 +234,10 @@ public class FtpSourceTests { 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(received, new Exception("just a test")); - assertEquals(received, ftpSource.receive()); + assertEquals(received.getPayload(), ftpSource.receive().getPayload()); verify(globalMocks); } diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/CustomMessageCreator.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/CustomMessageCreator.java deleted file mode 100644 index 829fa52267..0000000000 --- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/CustomMessageCreator.java +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Copyright 2002-2008 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.ftp.config; - -import java.io.File; - -import org.springframework.integration.message.GenericMessage; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageCreator; - -/** - * @author Marius Bogoevici - */ -public class CustomMessageCreator implements MessageCreator { - - public Message createMessage(File object) { - return new GenericMessage(object.getAbsolutePath()); - } - -} diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java index add55c2506..2e5b228a85 100644 --- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java +++ b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java @@ -17,7 +17,6 @@ package org.springframework.integration.ftp.config; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; import java.io.File; @@ -26,7 +25,6 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.message.DefaultMessageCreator; /** * @author Mark Fisher @@ -36,10 +34,10 @@ import org.springframework.integration.message.DefaultMessageCreator; public class FtpInboundChannelAdapterParserTests { @Test - public void ftpSourceWithDefaultMessageCreator() { + public void ftpInboundChannelAdapter() { ApplicationContext context = new ClassPathXmlApplicationContext( "ftpInboundChannelAdapterParserTests.xml", this.getClass()); - Object adapter = context.getBean("default.adapter"); + Object adapter = context.getBean("adapter"); DirectFieldAccessor sourceAccessor = new DirectFieldAccessor( new DirectFieldAccessor(adapter).getPropertyValue("source")); DirectFieldAccessor poolAccessor = new DirectFieldAccessor( @@ -50,27 +48,6 @@ public class FtpInboundChannelAdapterParserTests { assertEquals("/remote", poolAccessor.getPropertyValue("remoteWorkingDirectory")); assertEquals("testUser", poolAccessor.getPropertyValue("username")); assertEquals("testPassword", poolAccessor.getPropertyValue("password")); - Object messageCreator = sourceAccessor.getPropertyValue("messageCreator"); - assertTrue(messageCreator instanceof DefaultMessageCreator); - } - - @Test - public void ftpSourceWithCustomMessageCreator() { - ApplicationContext context = new ClassPathXmlApplicationContext( - "ftpInboundChannelAdapterParserTests.xml", this.getClass()); - Object adapter = context.getBean("custom.adapter"); - DirectFieldAccessor sourceAccessor = new DirectFieldAccessor( - new DirectFieldAccessor(adapter).getPropertyValue("source")); - DirectFieldAccessor poolAccessor = new DirectFieldAccessor( - sourceAccessor.getPropertyValue("clientPool")); - assertEquals("testHost", poolAccessor.getPropertyValue("host")); - assertEquals(2121, poolAccessor.getPropertyValue("port")); - assertEquals(new File("/local"), sourceAccessor.getPropertyValue("localWorkingDirectory")); - assertEquals("/remote", poolAccessor.getPropertyValue("remoteWorkingDirectory")); - assertEquals("testUser", poolAccessor.getPropertyValue("username")); - assertEquals("testPassword", poolAccessor.getPropertyValue("password")); - Object messageCreator = sourceAccessor.getPropertyValue("messageCreator"); - assertTrue(messageCreator instanceof CustomMessageCreator); } } diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSourceIntegrationTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSourceIntegrationTests.java index 5ac23e872d..54f3c9df69 100644 --- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSourceIntegrationTests.java +++ b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSourceIntegrationTests.java @@ -28,9 +28,7 @@ import org.junit.Test; import org.springframework.integration.ftp.FtpSource; import org.springframework.integration.ftp.QueuedFTPClientPool; -import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageCreator; /** * @author Iwein Fuld @@ -53,12 +51,6 @@ public class FtpSourceIntegrationTests { private FtpSource ftpSource; - private MessageCreator, List> messageCreator = new MessageCreator, List>() { - public Message> createMessage(List object) { - return new GenericMessage>(object); - } - }; - @BeforeClass public static void initializeEnvironment() { @@ -69,7 +61,7 @@ public class FtpSourceIntegrationTests { @Before public void initializeFtpSource() throws Exception { QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool(); - ftpSource = new FtpSource(messageCreator, queuedFTPClientPool); + ftpSource = new FtpSource(queuedFTPClientPool); queuedFTPClientPool.setHost("localhost"); queuedFTPClientPool.setUsername("ftp-user"); queuedFTPClientPool.setPassword("kaas"); diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/ftpInboundChannelAdapterParserTests.xml b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/ftpInboundChannelAdapterParserTests.xml index 8bfbcadd9d..527faabb22 100644 --- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/ftpInboundChannelAdapterParserTests.xml +++ b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/ftpInboundChannelAdapterParserTests.xml @@ -2,29 +2,23 @@ - + + - - - -