diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FTPClientFactory.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FTPClientFactory.java index 1e50192576..3da3e42273 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FTPClientFactory.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FTPClientFactory.java @@ -1,12 +1,36 @@ +/* + * 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.adapter.ftp; import java.io.IOException; import java.net.SocketException; import org.apache.commons.net.ftp.FTPClient; - +/** + * Factory for {@link FTPClient}. + * @author Iwein Fuld + * + */ public interface FTPClientFactory { + /** + * @return Fully configured and connected FTPClient. + * @throws SocketException + * @throws IOException + */ FTPClient getClient() throws SocketException, IOException; } 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 69d394dd75..2c7a8d623d 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,7 +38,7 @@ public class FtpTarget implements MessageTarget { private final MessageMapper messageMapper; - private volatile FTPClientPool ftpClientPool = new FTPClientPool(); + private volatile QueuedFTPClientPool ftpClientPool = new QueuedFTPClientPool(); public FtpTarget(MessageMapper messageMapper) { @@ -47,7 +47,7 @@ public class FtpTarget implements MessageTarget { } - public void setFtpClientPool(FTPClientPool ftpClientPool) { + public void setFtpClientPool(QueuedFTPClientPool ftpClientPool) { Assert.notNull(ftpClientPool, "ftpClientPool must not be null"); this.ftpClientPool = ftpClientPool; } 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 new file mode 100644 index 0000000000..5453257cf9 --- /dev/null +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPool.java @@ -0,0 +1,125 @@ +/* + * 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.adapter.ftp; + +import java.io.IOException; +import java.net.SocketException; +import java.util.Queue; +import java.util.concurrent.ArrayBlockingQueue; + +import org.apache.commons.logging.Log; +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.springframework.util.Assert; + +/** + * FTPClientPool implementation based on a Queue. This implementation has a + * default pool size of 5, but this is configurable with a constructor argument. + * + * @author Iwein Fuld + * + */ +public class QueuedFTPClientPool implements FTPClientPool { + + private static final int DEFAULT_POOL_SIZE = 5; + + private final Queue pool; + + private FTPClientConfig config; + + private String host; + + private int port = FTP.DEFAULT_PORT; + + private String user; + + private String pass; + + private FTPClientFactory factory = new DefaultFactory(); + + private final Log log = LogFactory.getLog(this.getClass()); + + public QueuedFTPClientPool() { + this(DEFAULT_POOL_SIZE); + } + + /** + * @param maxPoolSize the maximum size of the pool + */ + public QueuedFTPClientPool(int maxPoolSize) { + pool = new ArrayBlockingQueue(maxPoolSize); + } + + public synchronized FTPClient getClient() throws SocketException, IOException { + return pool.isEmpty() ? factory.getClient() : pool.poll(); + } + + public synchronized void releaseClient(FTPClient client) { + if (client != null) { + if (!pool.offer(client)) { + try { + client.disconnect(); + } + catch (IOException e) { + log.warn("Error disconnecting ftpclient", e); + } + } + } + } + + private class DefaultFactory implements FTPClientFactory { + + public FTPClient getClient() throws SocketException, IOException { + FTPClient client = new FTPClient(); + client.configure(config); + client.connect(host, port); + client.login(user, pass); + return client; + } + } + + public void setConfig(FTPClientConfig config) { + Assert.notNull(config); + this.config = config; + } + + public void setHost(String host) { + Assert.hasText(host); + this.host = host; + } + + public void setPort(int port) { + Assert.isTrue(port != 0, "Port number should be > 0"); + this.port = port; + } + + public void setUser(String user) { + Assert.hasText(user); + this.user = user; + } + + public void setPass(String pass) { + Assert.notNull(pass); + this.pass = pass; + } + + public void setFactory(FTPClientFactory factory) { + Assert.notNull(factory); + this.factory = factory; + } +} diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPoolTest.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPoolTest.java new file mode 100644 index 0000000000..c13085d3d6 --- /dev/null +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPoolTest.java @@ -0,0 +1,98 @@ +/* + * 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.adapter.ftp; + +import static junit.framework.Assert.assertEquals; +import static junit.framework.Assert.assertSame; +import static junit.framework.Assert.assertTrue; +import static org.easymock.EasyMock.createMock; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.replay; +import static org.easymock.EasyMock.verify; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.net.ftp.FTPClient; +import org.junit.Before; +import org.junit.Test; + +/** + * + * @author Iwein Fuld + * + */ +public class QueuedFTPClientPoolTest { + + private QueuedFTPClientPool pool; + + private FTPClientFactory factoryMock = createMock(FTPClientFactory.class); + + private Object[] allMocks = new Object[] { factoryMock }; + + @Before + public void initializeSubject() throws Exception { + this.pool = new QueuedFTPClientPool(5); + pool.setFactory(factoryMock); + } + + @Test + public void get() throws Exception { + FTPClient expectedClient = new FTPClient(); + expect(factoryMock.getClient()).andReturn(expectedClient); + replay(allMocks); + FTPClient client = pool.getClient(); + assertEquals(expectedClient, client); + verify(allMocks); + } + + @Test + public void getMultipleGet() throws Exception { + FTPClient[] expectedClients = new FTPClient[] { new FTPClient(), new FTPClient(), new FTPClient(), + new FTPClient(), new FTPClient(), new FTPClient() }; + for (FTPClient client : expectedClients) { + expect(factoryMock.getClient()).andReturn(client); + } + replay(allMocks); + for (int i = 0; i < 6; i++) { + assertSame(expectedClients[i], pool.getClient()); + } + verify(allMocks); + } + + @Test + public void getMultipleGetReleaseGet() throws Exception { + FTPClient[] expectedClients = new FTPClient[] { new FTPClient(), new FTPClient(), new FTPClient(), + new FTPClient(), new FTPClient() }; + for (FTPClient client : expectedClients) { + expect(factoryMock.getClient()).andReturn(client); + } + replay(allMocks); + List fromPool = new ArrayList(); + for (int i = 0; i < 5; i++) { + fromPool.add(pool.getClient()); + } + for (FTPClient client2 : fromPool) { + pool.releaseClient(client2); + } + for (int i = 0; i < 5; i++) { + FTPClient client = pool.getClient(); + boolean removed = fromPool.remove(client); + assertTrue("Failed on element " + i, removed); + } + verify(allMocks); + } +} 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 129bee4dd0..f00fe016c4 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,7 +25,7 @@ import org.junit.AfterClass; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; -import org.springframework.integration.adapter.ftp.FTPClientPool; +import org.springframework.integration.adapter.ftp.QueuedFTPClientPool; import org.springframework.integration.adapter.ftp.FtpTarget; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; @@ -46,7 +46,7 @@ public class FtpTargetIntegrationTest { return message.getPayload(); } }); - FTPClientPool clientPool = new FTPClientPool(); + QueuedFTPClientPool clientPool = new QueuedFTPClientPool(); clientPool.setHost("localhost"); clientPool.setUser("ftp-user"); clientPool.setPass("kaas");