Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Finishing up INT-293, INT-154. Parametrized DefaultMessageMapper, refactored FtpSource to use a pool, added namespace support for FtpTarget.

Browse Source
This commit is contained in:
Iwein Fuld
2008-08-17 05:37:03 +00:00
parent 8272a7d94c
commit d6aed95948
13 changed files with 211 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java
View File

@@ -30,7 +30,6 @@ 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.FTPFile;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.integration.adapter.file.AbstractDirectorySource;
import org.springframework.integration.adapter.file.Backlog;
@@ -48,56 +47,19 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
* @author Iwein Fuld
*/
public class FtpSource extends AbstractDirectorySource<List<File>> implements DisposableBean {
private final static String DEFAULT_HOST = "localhost";
private final static int DEFAULT_PORT = 21;
private final static String DEFAULT_REMOTE_WORKING_DIRECTORY = "/";
public class FtpSource extends AbstractDirectorySource<List<File>> {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile String username;
private volatile String password;
private volatile String host = DEFAULT_HOST;
private volatile int port = DEFAULT_PORT;
private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY;
private volatile File localWorkingDirectory;
private int maxFilesPerPayload = -1;
private final FTPClient client;
private final FTPClientPool clientPool;
public FtpSource(MessageCreator<List<File>, List<File>> messageCreator) {
this(messageCreator, new FTPClient());
}
public FtpSource(MessageCreator<List<File>, List<File>> messageCreator, FTPClient client) {
public FtpSource(MessageCreator<List<File>, List<File>> messageCreator, FTPClientPool clientPool) {
super(messageCreator);
this.client = client;
}
public void setHost(String host) {
this.host = host;
}
public void setPort(int port) {
this.port = port;
}
@Required
public void setUsername(String username) {
this.username = username;
}
public void setPassword(String password) {
this.password = password;
this.clientPool = clientPool;
}
public void setMaxMessagesPerPayload(int maxMessagesPerPayload) {
@@ -105,12 +67,6 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
this.maxFilesPerPayload = maxMessagesPerPayload;
}
public void setRemoteWorkingDirectory(String remoteWorkingDirectory) {
Assert.notNull(remoteWorkingDirectory, "'remoteWorkingDirectory' cannot be null");
// FtpClient is picky about "", so we make it happy
this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/");
}
public void setLocalWorkingDirectory(File localWorkingDirectory) {
Assert.notNull(localWorkingDirectory, "'localWorkingDirectory' must not be null");
this.localWorkingDirectory = localWorkingDirectory;
@@ -130,87 +86,48 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
@Override
protected void populateSnapshot(Map<String, FileInfo> snapshot) throws IOException {
establishConnection();
FTPFile[] fileList = this.client.listFiles();
for (FTPFile ftpFile : fileList) {
/*
* according to the FTPFile javadoc the list can contain nulls if
* files couldn't be parsed
*/
if (ftpFile != null) {
FileInfo fileInfo = new FileInfo(ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(), ftpFile
.getSize());
snapshot.put(ftpFile.getName(), fileInfo);
FTPClient client = clientPool.getClient();
FTPFile[] fileList = client.listFiles();
try {
for (FTPFile ftpFile : fileList) {
/*
* according to the FTPFile javadoc the list can contain nulls
* if files couldn't be parsed
*/
if (ftpFile != null) {
FileInfo fileInfo = new FileInfo(ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(),
ftpFile.getSize());
snapshot.put(ftpFile.getName(), fileInfo);
}
}
}
}
protected void establishConnection() throws IOException {
if (this.client.isConnected()) {
if (logger.isDebugEnabled()) {
logger.debug("client already connected");
}
return;
}
if (!StringUtils.hasText(this.username)) {
throw new MessagingException("username is required");
}
this.client.connect(this.host, this.port);
if (!this.client.login(this.username, this.password)) {
throw new MessagingException("Login failed. Please check the username and password.");
}
if (logger.isDebugEnabled()) {
logger.debug("login successful");
}
this.client.setFileType(FTP.IMAGE_FILE_TYPE);
if (!this.remoteWorkingDirectory.equals(this.client.printWorkingDirectory())
&& !this.client.changeWorkingDirectory(this.remoteWorkingDirectory)) {
throw new MessagingException("Could not change directory to '" + remoteWorkingDirectory
+ "'. Please check the path.");
}
if (logger.isDebugEnabled()) {
logger.debug("working directory is: " + this.client.printWorkingDirectory());
finally {
clientPool.releaseClient(client);
}
}
protected List<File> retrieveNextPayload() throws IOException {
establishConnection();
List<File> files = new ArrayList<File>();
Set<String> toDo = this.getDirectoryContentManager().getProcessingBuffer().keySet();
for (String fileName : toDo) {
File file = new File(this.localWorkingDirectory, fileName);
if (file.exists()) {
file.delete();
}
FileOutputStream fileOutputStream = new FileOutputStream(file);
this.client.retrieveFile(fileName, fileOutputStream);
fileOutputStream.close();
files.add(file);
}
disconnect();
return files;
}
protected void disconnect() {
FTPClient client = clientPool.getClient();
try {
if (this.client.isConnected()) {
this.client.disconnect();
if (logger.isDebugEnabled()) {
logger.debug("connection closed");
List<File> files = new ArrayList<File>();
Set<String> toDo = this.getDirectoryContentManager().getProcessingBuffer().keySet();
for (String fileName : toDo) {
File file = new File(this.localWorkingDirectory, fileName);
if (file.exists()) {
file.delete();
}
FileOutputStream fileOutputStream = new FileOutputStream(file);
client.retrieveFile(fileName, fileOutputStream);
fileOutputStream.close();
files.add(file);
}
return files;
}
catch (IOException ioe) {
if (logger.isErrorEnabled()) {
logger.error("Error when disconnecting from ftp.", ioe);
}
finally {
clientPool.releaseClient(client);
}
}
public void destroy() throws Exception {
disconnect();
}
@SuppressWarnings("unchecked")
@Override

12
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpTarget.java
View File

@@ -38,18 +38,14 @@ public class FtpTarget implements MessageTarget {
private final MessageMapper<?, File> messageMapper;
private volatile FTPClientPool ftpClientPool = new QueuedFTPClientPool();
private final FTPClientPool ftpClientPool;
public FtpTarget(MessageMapper<?, File> messageMapper) {
Assert.notNull(messageMapper, "MessageMapper must not be null");
this.messageMapper = messageMapper;
}
public void setFtpClientPool(FTPClientPool ftpClientPool) {
public FtpTarget(MessageMapper<?, File> messageMapper, FTPClientPool ftpClientPool) {
Assert.notNull(messageMapper, "messageMapper must not be null");
Assert.notNull(ftpClientPool, "ftpClientPool must not be null");
this.ftpClientPool = ftpClientPool;
this.messageMapper = messageMapper;
}
public boolean send(Message message) {

51
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPool.java
View File

@@ -26,8 +26,11 @@ 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.apache.commons.net.ftp.FTPReply;
import org.springframework.integration.message.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* FTPClientPool implementation based on a Queue. This implementation has a
@@ -39,6 +42,8 @@ public class QueuedFTPClientPool implements FTPClientPool {
private static final int DEFAULT_POOL_SIZE = 5;
private static final String DEFAULT_REMOTE_WORKING_DIRECTORY = "/";
private final Queue<FTPClient> pool;
private volatile FTPClientConfig config;
@@ -47,14 +52,16 @@ public class QueuedFTPClientPool implements FTPClientPool {
private volatile int port = FTP.DEFAULT_PORT;
private volatile String user;
private volatile String username;
private volatile String pass;
private volatile String password;
private volatile FTPClientFactory factory = new DefaultFactory();
private final Log log = LogFactory.getLog(this.getClass());
private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY;
public QueuedFTPClientPool() {
this(DEFAULT_POOL_SIZE);
}
@@ -98,14 +105,19 @@ public class QueuedFTPClientPool implements FTPClientPool {
this.port = port;
}
public void setUser(String user) {
public void setUsername(String user) {
Assert.hasText(user);
this.user = user;
this.username = user;
}
public void setPass(String pass) {
public void setPassword(String pass) {
Assert.notNull(pass);
this.pass = pass;
this.password = pass;
}
public void setRemoteWorkingDirectory(String remoteWorkingDirectory) {
Assert.notNull(remoteWorkingDirectory);
this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/");
}
public void setFactory(FTPClientFactory factory) {
@@ -118,8 +130,33 @@ public class QueuedFTPClientPool implements FTPClientPool {
public FTPClient getClient() throws SocketException, IOException {
FTPClient client = new FTPClient();
client.configure(config);
if (!StringUtils.hasText(username)) {
throw new MessagingException("username is required");
}
client.connect(host, port);
client.login(user, pass);
if (!FTPReply.isPositiveCompletion(client.getReplyCode())) {
throw new MessagingException("Connecting to server [" + host + ":" + port
+ "] failed, please check the connection");
}
if (log.isDebugEnabled()) {
log.debug("Connected to server [" + host + ":" + port + "]");
}
if (!client.login(username, password)) {
throw new MessagingException("Login failed. Please check the username and password.");
}
if (log.isDebugEnabled()) {
log.debug("login successful");
}
client.setFileType(FTP.BINARY_FILE_TYPE);
if (!remoteWorkingDirectory.equals(client.printWorkingDirectory())
&& !client.changeWorkingDirectory(remoteWorkingDirectory)) {
throw new MessagingException("Could not change directory to '" + remoteWorkingDirectory
+ "'. Please check the path.");
}
if (log.isDebugEnabled()) {
log.debug("working directory is: " + client.printWorkingDirectory());
}
return client;
}
}

42
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java
View File

@@ -16,27 +16,63 @@
package org.springframework.integration.adapter.ftp.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.integration.adapter.file.config.AbstractDirectorySourceParser;
import org.springframework.integration.adapter.ftp.FtpSource;
import org.springframework.integration.adapter.ftp.QueuedFTPClientPool;
import org.w3c.dom.Element;
/**
* Parser for the &lt;ftp-source/&gt; element.
*
* @author Mark Fisher
* @author Marius Bogoevici
* @author Iwein Fuld
*/
public class FtpSourceParser extends AbstractDirectorySourceParser {
private static final String POOL_ATTRIBUTE_USER = "username";
private static final String POOL_ATTRIBUTE_PASS = "password";
private static final String POOL_ATTRIBUTE_HOST = "host";
private static final String POOL_ATTRIBUTE_PORT = "port";
private static final String POOL_ATTRIBUTE_REMOTEDIR = "remote-working-directory";
public FtpSourceParser() {
super(true);
}
@Override
protected Class<?> getBeanClass(Element element) {
return FtpSource.class;
}
@Override
protected boolean isEligibleAttribute(String attributeName) {
return !POOL_ATTRIBUTE_HOST.equals(attributeName)
&& !POOL_ATTRIBUTE_PASS.equals(attributeName)
&& !POOL_ATTRIBUTE_PORT.equals(attributeName)
&& !POOL_ATTRIBUTE_USER.equals(attributeName)
&& !POOL_ATTRIBUTE_REMOTEDIR.equals(attributeName)
&& super.isEligibleAttribute(attributeName);
}
@Override
protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) {
super.postProcess(beanDefinition, element);
String user = element.getAttribute(POOL_ATTRIBUTE_USER);
String pass = element.getAttribute(POOL_ATTRIBUTE_PASS);
String host = element.getAttribute(POOL_ATTRIBUTE_HOST);
String port = element.getAttribute(POOL_ATTRIBUTE_PORT);
String remoteWorkingDirectory = element.getAttribute(POOL_ATTRIBUTE_REMOTEDIR);
QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool();
queuedFTPClientPool.setUsername(user);
queuedFTPClientPool.setPassword(pass);
queuedFTPClientPool.setHost(host);
queuedFTPClientPool.setPort(Integer.parseInt(port));
queuedFTPClientPool.setRemoteWorkingDirectory(remoteWorkingDirectory);
beanDefinition.addConstructorArgValue(queuedFTPClientPool);
}
}