Added noOp check to QueuedFtpClientPool

removed tracing from PollableFileSource

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileTarget.java

@@ -24,21 +24,19 @@ import org.springframework.integration.message.MessageTarget;
 
 /**
  * A message target for writing files. The actual file writing occurs in the
- * message creator ({@link TextFileMessageCreator} or {@link ByteArrayFileMessageCreator}).
+ * message creator ({@link TextFileMessageCreator} or
+ * {@link ByteArrayFileMessageCreator}).
  * 
  * @author Mark Fisher
  * @author Marius Bogoevici
  */
 public class FileTarget implements MessageTarget {
-
 	private MessageMapper<?, File> messageMapper;
 
-
 	public FileTarget(MessageMapper<?, File> messageMapper) {
 		this.messageMapper = messageMapper;
 	}
 
-
 	public boolean send(Message message) {
 		File file = this.messageMapper.mapMessage(message);
 		return file.exists();

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPool.java

@@ -36,6 +36,9 @@ import org.springframework.util.StringUtils;
  * FTPClientPool implementation based on a Queue. This implementation has a
  * default pool size of 5, but this is configurable with a constructor argument.
  * 
+ * This implementation pools released clients, but gives no guarantee to the
+ * number of clients open at the same time.
+ * 
  * @author Iwein Fuld
  */
 public class QueuedFTPClientPool implements FTPClientPool {
@@ -62,34 +65,7 @@ public class QueuedFTPClientPool implements FTPClientPool {
 
 	private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY;
 
-	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 && client.isConnected()) {
-			if (!pool.offer(client)) {
-				try {
-					client.disconnect();
-				}
-				catch (IOException e) {
-					log.warn("Error disconnecting ftpclient", e);
-				}
-			}
-		}
-	}
-
+	// setters
 	public void setConfig(FTPClientConfig config) {
 		Assert.notNull(config);
 		this.config = config;
@@ -106,17 +82,17 @@ public class QueuedFTPClientPool implements FTPClientPool {
 	}
 
 	public void setUsername(String user) {
-		Assert.hasText(user);
+		Assert.hasText(user, "'user' should be a nonempty string");
 		this.username = user;
 	}
 
 	public void setPassword(String pass) {
-		Assert.notNull(pass);
+		Assert.notNull(pass, "password should not be null");
 		this.password = pass;
 	}
 
 	public void setRemoteWorkingDirectory(String remoteWorkingDirectory) {
-		Assert.notNull(remoteWorkingDirectory);
+		Assert.notNull(remoteWorkingDirectory, "remote directory should not be null");
 		this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/");
 	}
 
@@ -125,6 +101,62 @@ 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<FTPClient>(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
+	 * reason large pools are not recommended in poor networking conditions.
+	 */
+	public FTPClient getClient() throws SocketException, IOException {
+		FTPClient client = pool.poll();
+		if (client == null) {
+			client = factory.getClient();
+		}
+		else {
+			client = isClientAlive(client) ? client : getClient();
+		}
+		return client;
+	}
+
+	private boolean isClientAlive(FTPClient client) {
+		try {
+			if (client.sendNoOp()) {
+				return true;
+			}
+		}
+		catch (IOException e) {
+			log.warn("Client [" + client + "] discarded: ", e);
+		}
+		return false;
+	}
+
+	public void releaseClient(FTPClient client) {
+		Assert.notNull(client, "'client' cannot be 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 {
@@ -160,5 +192,4 @@ public class QueuedFTPClientPool implements FTPClientPool {
 			return client;
 		}
 	}
-
 }

org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPoolTest.java

@@ -18,14 +18,16 @@ 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.*;
+import static org.easymock.classextension.EasyMock.*;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.commons.net.ftp.FTPClient;
 import org.junit.Before;
 import org.junit.Test;
+import org.springframework.integration.adapter.file.FileTarget;
 
 /**
  * 
@@ -58,8 +60,8 @@ public class QueuedFTPClientPoolTest {
 
 	@Test
 	public void getMultipleGet() throws Exception {
-		FTPClient[] expectedClients = new FTPClient[] { connectedFTPClient(), connectedFTPClient(),
-				connectedFTPClient(), connectedFTPClient(), connectedFTPClient(), connectedFTPClient() };
+		FTPClient[] expectedClients = new FTPClient[] { mockedFTPClient(), mockedFTPClient(),
+				mockedFTPClient(), mockedFTPClient(), mockedFTPClient(), mockedFTPClient() };
 		for (FTPClient client : expectedClients) {
 			expect(factoryMock.getClient()).andReturn(client);
 		}
@@ -72,8 +74,8 @@ public class QueuedFTPClientPoolTest {
 
 	@Test
 	public void getMultipleGetReleaseGet() throws Exception {
-		FTPClient[] expectedClients = new FTPClient[] { connectedFTPClient(), connectedFTPClient(),
-				connectedFTPClient(), connectedFTPClient(), connectedFTPClient() };
+		FTPClient[] expectedClients = new FTPClient[] { mockedFTPClient(), mockedFTPClient(),
+				mockedFTPClient(), mockedFTPClient(), mockedFTPClient() };
 		for (FTPClient client : expectedClients) {
 			expect(factoryMock.getClient()).andReturn(client);
 		}
@@ -93,12 +95,11 @@ public class QueuedFTPClientPoolTest {
 		verify(allMocks);
 	}
 
-	private FTPClient connectedFTPClient() {
-		return new FTPClient() {
-			@Override
-			public boolean isConnected() {
-				return true;
-			}
-		};
+	private FTPClient mockedFTPClient() throws Exception {
+		FTPClient mock = createNiceMock(FTPClient.class);
+		expect(mock.isConnected()).andReturn(true).anyTimes();
+		expect(mock.sendNoOp()).andReturn(true).anyTimes();
+		replay(mock);
+		return mock;
 	}
 }