diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileTarget.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileTarget.java index 2630f11211..3ab722c4a4 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileTarget.java +++ b/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 messageMapper; - public FileTarget(MessageMapper messageMapper) { this.messageMapper = messageMapper; } - public boolean send(Message message) { File file = this.messageMapper.mapMessage(message); return file.exists(); 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 0689b9a069..8212f1a37f 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 @@ -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(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(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; } } - } 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 index a6f3783e8b..98420dc17b 100644 --- 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 @@ -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; } } diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.java index 7b89a318d8..b9fa4ff8d3 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/PollableFileSource.java @@ -82,7 +82,6 @@ public class PollableFileSource implements PollableSource, MessageDelivery } public Message receive() throws MessagingException { - traceState(); refreshQueue(); Message message = null; File file = toBeReceived.poll(); @@ -93,7 +92,6 @@ public class PollableFileSource implements PollableSource, MessageDelivery logger.info("Created message: [" + message + "]"); } } - traceState(); return message; } @@ -127,13 +125,4 @@ public class PollableFileSource implements PollableSource, MessageDelivery logger.debug("Sent: " + sentMessage); } } - - /* - * utility method to trace the stateful collections of this instance. - */ - private void traceState() { - if (logger.isTraceEnabled()) { - logger.trace("Files to be received: [" + toBeReceived + "]"); - } - } }