FTPClientPool improvements and javadoc.

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;
 
 }

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<?, File> messageMapper;
 
-	private volatile FTPClientPool ftpClientPool = new FTPClientPool();
+	private volatile QueuedFTPClientPool ftpClientPool = new QueuedFTPClientPool();
 
 
 	public FtpTarget(MessageMapper<?, File> 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;
 	}

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<FTPClient> 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<FTPClient>(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;
+	}
+}

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<FTPClient> fromPool = new ArrayList<FTPClient>();
+		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);
+	}
+}

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");