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

moving FTP / SFTP support from sandbox.

Browse Source
This commit is contained in:
Josh Long
2010-08-17 03:59:52 +00:00
parent 57d5e5c400
commit d85fdaceb0
44 changed files with 3569 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pom.xml
View File

@@ -28,6 +28,8 @@
<module>spring-integration-ws</module>
<module>spring-integration-xml</module>
<module>spring-integration-xmpp</module>
<module>spring-integration-ftp</module>
<module>spring-integration-sftp</module>
</modules>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

99
spring-integration-ftp/pom.xml Normal file
View File

@@ -0,0 +1,99 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-parent</artifactId>
<version>2.0.0.BUILD-SNAPSHOT</version>
</parent>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-ftp</artifactId>
<packaging>jar</packaging>
<name>Spring Integration FTP Support</name>
<dependencies>
<dependency>
<groupId>javax.activation</groupId>
<artifactId>activation</artifactId>
<version>1.1.1</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-file</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>commons-net</groupId>
<artifactId>commons-net</artifactId>
<version>2.0</version>
</dependency>
<dependency>
<groupId>cglib</groupId>
<artifactId>cglib-nodep</artifactId>
<version>${cglib.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.easymock</groupId>
<artifactId>easymock</artifactId>
<version>${org.easymock.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.easymock</groupId>
<artifactId>easymockclassextension</artifactId>
<version>${org.easymock.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>${junit.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context-support</artifactId>
<version>${org.springframework.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${org.springframework.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
<version>${project.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>${project.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.5</version>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>1.4</version>
</dependency>
</dependencies>
<build>
<!-- <plugins>
<plugin>
<groupId>com.springsource.bundlor</groupId>
<artifactId>com.springsource.bundlor.maven</artifactId>
</plugin>
</plugins> -->
</build>
</project>

146
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/DefaultFTPClientFactory.java Normal file
View File

@@ -0,0 +1,146 @@
/*
* 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.ftp;
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.apache.commons.net.ftp.FTPReply;
import org.springframework.integration.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import java.io.IOException;
import java.net.SocketException;
/**
* Default implementation of FTPClientFactory.
*
* @author iwein
*/
public class DefaultFTPClientFactory implements FTPClientFactory {
private static final Log logger = LogFactory.getLog(FTPClientFactory.class);
private static final String DEFAULT_REMOTE_WORKING_DIRECTORY = "/";
private FTPClientConfig config;
private String username;
private String host;
private int port = FTP.DEFAULT_PORT;
private String password;
private String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY;
private int clientMode = FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE;
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 setUsername(String user) {
Assert.hasText(user, "'user' should be a nonempty string");
this.username = user;
}
public void setPassword(String pass) {
Assert.notNull(pass, "password should not be null");
this.password = pass;
}
public void setRemoteWorkingDirectory(String remoteWorkingDirectory) {
Assert.notNull(remoteWorkingDirectory, "remote directory should not be null");
this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/");
}
/**
* Set client mode for example
* <code>FTPClient.ACTIVE_LOCAL_CONNECTION_MODE</code> (default) Only local
* modes are supported.
*/
public void setClientMode(int clientMode) {
this.clientMode = clientMode;
}
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);
setClientMode(client);
if (!FTPReply.isPositiveCompletion(client.getReplyCode())) {
throw new MessagingException("Connecting to server [" + host + ":" + port + "] failed, please check the connection");
}
if (logger.isDebugEnabled()) {
logger.debug("Connected to server [" + host + ":" + port + "]");
}
if (!client.login(username, password)) {
throw new MessagingException("Login failed. Please check the username and password.");
}
if (logger.isDebugEnabled()) {
logger.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 (logger.isDebugEnabled()) {
logger.debug("working directory is: " + client.printWorkingDirectory());
}
return client;
}
/**
* Sets the mode of the connection. Only local modes are supported.
*/
private void setClientMode(FTPClient client) {
switch (clientMode) {
case FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE:
client.enterLocalActiveMode();
break;
case FTPClient.PASSIVE_LOCAL_DATA_CONNECTION_MODE:
client.enterLocalPassiveMode();
break;
default:
break;
}
}
}

34
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPClientFactory.java Normal file
View File

@@ -0,0 +1,34 @@
/*
* 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.ftp;
import org.apache.commons.net.ftp.FTPClient;
import java.io.IOException;
/**
* Factory for {@link FTPClient}.
*
* @author Iwein Fuld
*/
public interface FTPClientFactory {
/**
* @return Fully configured and connected FTPClient. Never <code>null</code>.
* @throws IOException thrown when a networking IO subsystem error occurs
*/
FTPClient getClient() throws IOException;
}

46
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPClientPool.java Normal file
View File

@@ -0,0 +1,46 @@
/*
* 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.ftp;
import org.apache.commons.net.ftp.FTPClient;
/**
* A pool of {@link FTPClient} instances. The pool can be used to control the
* number of open FTP connections and reuse these connections efficiently.
*
* @author Iwein Fuld
*/
public interface FTPClientPool extends FTPClientFactory {
/**
* Releases the client back to the pool. When calling this method the caller
* is no longer responsible for the connection. The pool is free to do with
* it as it sees fit, which means either recycling or disconnecting it most
* probably.
* <p/>
* The caller should NOT disconnect the client before calling this method.
* <p/>
* The caller is NOT expected to use the client after calling this method.
* Doing so can lead to unexpected behavior.
*
* @param client the {@link FTPClient} to release. Implementations of this
* method are recommended to deal gracefully with a <code>null</code>
* argument, although the endpoint implementations in
* <code>org.springframework.integration.ftp</code> will never pass in
* <code>null</code>.
*/
void releaseClient(FTPClient client);
}

114
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileSource.java Normal file
View File

@@ -0,0 +1,114 @@
/*
* 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.ftp;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.io.Resource;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.AcceptOnceFileListFilter;
import org.springframework.integration.file.CompositeFileListFilter;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.PatternMatchingFileListFilter;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import java.io.File;
import java.io.IOException;
import java.util.logging.Logger;
import java.util.regex.Pattern;
/**
* A source adapter for receiving files via FTP.
*
* @author Iwein Fuld
*/
public class FTPFileSource implements MessageSource<File>, InitializingBean, Lifecycle {
private static final Logger logger = Logger.getLogger(FTPFileSource.class.getName());
private FileReadingMessageSource fileSource;
private FTPInboundSynchronizer synchronizer;
public FTPFileSource() {
this(new FileReadingMessageSource(), new FTPInboundSynchronizer());
}
public FTPFileSource(FileReadingMessageSource fileSource, FTPInboundSynchronizer synchronizer) {
this.fileSource = fileSource;
this.synchronizer = synchronizer;
Pattern completePattern = Pattern.compile("^.*(?<!" + FTPInboundSynchronizer.INCOMPLETE_EXTENSION + ")$");
fileSource.setFilter(new CompositeFileListFilter(new AcceptOnceFileListFilter(), new PatternMatchingFileListFilter(completePattern)));
}
public void setFileSource(FileReadingMessageSource fileSource) {
this.fileSource = fileSource;
}
public void setSynchronizer(FTPInboundSynchronizer synchronizer) {
this.synchronizer = synchronizer;
}
public void setLocalWorkingDirectory(Resource localWorkingDirectory) {
this.synchronizer.setLocalDirectory(localWorkingDirectory);
try {
this.fileSource.setDirectory(localWorkingDirectory.getFile());
} catch (IOException e) {
}
}
public void setTrigger(Trigger trigger) {
synchronizer.setTrigger(trigger);
}
public void setTaskScheduler(TaskScheduler scheduler) {
synchronizer.setTaskScheduler(scheduler);
}
public void setClientPool(FTPClientPool pool) {
synchronizer.setClientPool(pool);
}
public void afterPropertiesSet() throws Exception {
synchronizer.afterPropertiesSet();
}
public Message<File> receive() {
return fileSource.receive();
}
public void onFailure(Message<File> failedMessage, Throwable t) {
fileSource.onFailure(failedMessage, t);
}
public void onSend(Message<File> sentMessage) {
fileSource.onSend(sentMessage);
}
public boolean isRunning() {
return synchronizer.isRunning();
}
public void start() {
synchronizer.start();
}
public void stop() {
synchronizer.stop();
}
}

165
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPInboundSynchronizer.java Normal file
View File

@@ -0,0 +1,165 @@
/*
* 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.ftp;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.io.Resource;
import org.springframework.integration.MessagingException;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.ScheduledFuture;
/**
* <code>FTPInboundSynchronizer</code> will keep a local directory in sync with a remote Ftp directory.
* It will NOT move new files put into the local directory to the remote server.
*
* @author Iwein Fuld
*
*/
public class FTPInboundSynchronizer implements InitializingBean, Lifecycle {
private static final Log logger = LogFactory.getLog(FTPInboundSynchronizer.class);
static final String INCOMPLETE_EXTENSION = ".INCOMPLETE";
private static final long DEFAULT_REFRESH_RATE = 10000;
private volatile TaskScheduler taskScheduler;
private volatile Trigger trigger = new PeriodicTrigger(DEFAULT_REFRESH_RATE);
private volatile FTPClientPool clientPool;
private volatile Resource localDirectory;
private boolean running = false;
private ScheduledFuture<?> scheduledFuture;
public void setTaskScheduler(TaskScheduler scheduler) {
this.taskScheduler = scheduler;
}
public void setTrigger(Trigger trigger) {
this.trigger = trigger;
}
public void setLocalDirectory(Resource localDirectory) {
this.localDirectory = localDirectory;
}
public void setClientPool(FTPClientPool pool) {
clientPool = pool;
}
public void afterPropertiesSet() throws Exception {
Assert.notNull(localDirectory, "'localDirectory' is required.");
}
private void synchronize() {
try {
FTPClient client = this.clientPool.getClient();
Assert.state(client != null, FTPClientPool.class.getSimpleName() + " returned 'null' client this most likely a bug in the pool implementation.");
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) && ftpFile.isFile()) {
copyFileToLocalDirectory(client, ftpFile, this.localDirectory);
}
}
} finally {
if (client != null) {
this.clientPool.releaseClient(client);
}
}
} catch (IOException e) {
throw new MessagingException("Problem occurred while synchronizing remote to local directory", e);
}
}
private boolean copyFileToLocalDirectory(FTPClient client, FTPFile ftpFile, Resource localDirectory)
throws IOException, FileNotFoundException {
String remoteFileName = ftpFile.getName();
String localFileName = localDirectory.getFile().getPath() + "/" + remoteFileName;
File localFile = new File(localFileName);
if (!localFile.exists()) {
String tempFileName = localFileName + INCOMPLETE_EXTENSION;
File file = new File(tempFileName);
FileOutputStream fos = new FileOutputStream(file);
try {
client.retrieveFile(remoteFileName, fos);
} finally {
fos.close();
}
file.renameTo(localFile);
return true;
} else {
return false;
}
}
public boolean isRunning() {
return running;
}
public void start() {
if (running) {
return;
}
Assert.state(taskScheduler != null, "'taskScheduler' is required");
scheduledFuture = taskScheduler.schedule(new SynchronizeTask(), trigger);
// future.get();
this.running = true;
if (logger.isInfoEnabled()) {
logger.info("Started " + this);
}
}
public void stop() {
if (!running) {
return;
}
this.scheduledFuture.cancel(true);
this.running = false;
if (logger.isInfoEnabled()) {
logger.info("Stopped " + this);
}
}
private class SynchronizeTask implements Runnable {
public void run() {
synchronize();
}
}
}

168
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPMessageSourceFactoryBean.java Normal file
View File

@@ -0,0 +1,168 @@
package org.springframework.integration.ftp;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.SystemUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceEditor;
import org.springframework.core.io.ResourceLoader;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ErrorHandler;
import java.io.File;
import java.util.Map;
/**
* Makes it easier to assemble the moving pieces involved in standing up an {@link org.springframework.integration.ftp.FTPMessageSourceFactoryBean}
*
* @author Josh Long
*/
public class FTPMessageSourceFactoryBean extends AbstractFactoryBean<FTPFileSource> implements ResourceLoaderAware, ApplicationContextAware {
private int port;
private boolean autoCreateDirectories;
private String username;
private String password;
private String host;
private String remoteDirectory;
private String localWorkingDirectory;
private ApplicationContext applicationContext;
private Resource localDirectoryResource;
private FTPInboundSynchronizer ftpInboundSynchronizer;
private TaskScheduler taskScheduler;
private ResourceLoader resourceLoader;
private FileReadingMessageSource fileReadingMessageSource;
private int clientMode = FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE;
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = applicationContext;
}
public void setPort(int port) {
this.port = port;
}
public void setPassword(String password) {
this.password = password;
}
public void setUsername(String username) {
this.username = username;
}
public void setAutoCreateDirectories(boolean autoCreateDirectories) {
this.autoCreateDirectories = autoCreateDirectories;
}
public void setRemoteDirectory(String remoteDirectory) {
this.remoteDirectory = remoteDirectory;
}
public void setHost(String host) {
this.host = host;
}
public void setLocalWorkingDirectory(String localWorkingDirectory) {
this.localWorkingDirectory = localWorkingDirectory;
}
public void setResourceLoader(ResourceLoader resourceLoader) {
this.resourceLoader = resourceLoader;
}
public Class<?extends FTPFileSource> getObjectType() {
return FTPFileSource.class;
}
public void setClientMode(int clientMode) {
this.clientMode = clientMode;
}
@Override
protected FTPFileSource createInstance() throws Exception {
// setup local dir
if (StringUtils.isEmpty(this.localWorkingDirectory)) {
File tmp = SystemUtils.getJavaIoTmpDir();
File ftpTmp = new File(tmp, "ftpInbound");
this.localWorkingDirectory = "file://" + ftpTmp.getAbsolutePath();
}
assert !StringUtils.isEmpty(this.localWorkingDirectory) : "the local working directory can't be null!";
ResourceEditor resourceEditor = new ResourceEditor(this.resourceLoader);
resourceEditor.setAsText(this.localWorkingDirectory);
this.localDirectoryResource = (Resource) resourceEditor.getValue();
fileReadingMessageSource = new FileReadingMessageSource();
this.ftpInboundSynchronizer = new FTPInboundSynchronizer();
if (this.taskScheduler == null) {
Map<String, TaskScheduler> tss = null;
if ((tss = applicationContext.getBeansOfType(TaskScheduler.class)).keySet().size() != 0) {
taskScheduler = tss.get(tss.keySet().iterator().next());
}
}
if (null == taskScheduler) {
ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
threadPoolTaskScheduler.setPoolSize(10);
threadPoolTaskScheduler.setErrorHandler(new ErrorHandler() {
public void handleError(Throwable t) {
logger.debug("Error! ", t);
}
});
threadPoolTaskScheduler.setWaitForTasksToCompleteOnShutdown(true);
threadPoolTaskScheduler.initialize();
this.taskScheduler = threadPoolTaskScheduler;
}
DefaultFTPClientFactory defaultFTPClientFactory = new DefaultFTPClientFactory();
defaultFTPClientFactory.setHost(this.host);
defaultFTPClientFactory.setPassword(this.password);
defaultFTPClientFactory.setPort(this.port);
defaultFTPClientFactory.setRemoteWorkingDirectory(this.remoteDirectory);
defaultFTPClientFactory.setUsername(this.username);
defaultFTPClientFactory.setClientMode(this.clientMode);
QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool(15, defaultFTPClientFactory);
this.ftpInboundSynchronizer.setClientPool(queuedFTPClientPool);
this.ftpInboundSynchronizer.setLocalDirectory(this.localDirectoryResource);
this.ftpInboundSynchronizer.setTaskScheduler(this.taskScheduler);
assert this.localDirectoryResource != null : "the 'localDirectoryResource' can't be null at this point";
if (this.autoCreateDirectories) {
if (!this.localDirectoryResource.exists()) {
try {
if (!localDirectoryResource.getFile().mkdirs()) {
logger.debug("attempted to ensure the existence of the local directory '" + this.localDirectoryResource.getFile().getAbsolutePath() + "' but didn't succeed");
}
} catch (Throwable th) {
logger.debug("attempted to ensure the existence of the local directory '" + this.localDirectoryResource.getFile().getAbsolutePath() + "' but didn't succeed");
}
}
}
this.ftpInboundSynchronizer.afterPropertiesSet();
FTPFileSource ftpFileSource = new FTPFileSource(this.fileReadingMessageSource, this.ftpInboundSynchronizer);
ftpFileSource.setClientPool(queuedFTPClientPool);
ftpFileSource.setLocalWorkingDirectory(this.localDirectoryResource);
ftpFileSource.setSynchronizer(this.ftpInboundSynchronizer);
ftpFileSource.setTaskScheduler(this.taskScheduler);
ftpFileSource.setFileSource(this.fileReadingMessageSource);
ftpFileSource.afterPropertiesSet();
ftpFileSource.start();
return ftpFileSource;
}
}

105
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPSendingMessageHandler.java Normal file
View File

@@ -0,0 +1,105 @@
/*
* 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.ftp;
import org.apache.commons.net.ftp.FTPClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.Message;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.util.Assert;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.SocketException;
/**
* A {@link org.springframework.integration.core.MessageHandler} implementation that sends files to an FTP server.
*
* @author Iwein Fuld
* @author Mark Fisher
*/
public class FTPSendingMessageHandler implements MessageHandler, InitializingBean {
private FTPClientPool ftpClientPool;
public FTPSendingMessageHandler() {
}
public FTPSendingMessageHandler(FTPClientPool ftpClientPool) {
this.ftpClientPool = ftpClientPool;
}
public void setFtpClientPool(FTPClientPool ftpClientPool) {
this.ftpClientPool = ftpClientPool;
}
public void afterPropertiesSet() throws Exception {
Assert.notNull(ftpClientPool, "'ftpClientPool' must not be null");
}
public void handleMessage(Message<?> message) {
Assert.notNull(message, "'message' must not be null");
Object payload = message.getPayload();
Assert.notNull(payload, "Message payload must not be null");
Assert.isInstanceOf(File.class, payload, "Message payload must be an instance of [java.io.File]");
File file = (File) payload;
if ((file != null) && file.exists()) {
FTPClient client = null;
boolean sentSuccesfully;
try {
client = getFtpClient();
sentSuccesfully = sendFile(file, client);
} catch (FileNotFoundException e) {
throw new MessageDeliveryException(message, "File [" + file + "] not found in local working directory; it was moved or deleted unexpectedly", e);
} catch (IOException e) {
throw new MessageDeliveryException(message, "Error transferring file [" + file + "] from local working directory to remote FTP directory", e);
} catch (Exception e) {
throw new MessageDeliveryException(message, "Error handling message for file [" + file + "]", e);
} finally {
if (client != null) {
ftpClientPool.releaseClient(client);
}
}
if (!sentSuccesfully) {
throw new MessageDeliveryException(message, "Failed to store file '" + file + "'");
}
}
}
private boolean sendFile(File file, FTPClient client)
throws FileNotFoundException, IOException {
FileInputStream fileInputStream = new FileInputStream(file);
boolean sent = client.storeFile(file.getName(), fileInputStream);
fileInputStream.close();
return sent;
}
private FTPClient getFtpClient() throws SocketException, IOException {
FTPClient client;
client = this.ftpClientPool.getClient();
Assert.state(client != null, FTPClientPool.class.getSimpleName() + " returned 'null' client this most likely a bug in the pool implementation.");
return client;
}
}

88
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPSendingMessageHandlerFactoryBean.java Normal file
View File

@@ -0,0 +1,88 @@
package org.springframework.integration.ftp;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.core.io.ResourceLoader;
/**
* A factory bean implementation that handles constructing an outbound FTP adapter.
*
* @author Josh Long
*/
public class FTPSendingMessageHandlerFactoryBean extends AbstractFactoryBean<FTPSendingMessageHandler> implements ResourceLoaderAware, ApplicationContextAware {
private int port;
private String username;
private String password;
private String host;
private String remoteDirectory;
private int clientMode;
// private vars
private ResourceLoader resourceLoader;
private ApplicationContext applicationContext;
public void setClientMode(int clientMode) {
this.clientMode = clientMode;
}
public void setResourceLoader(ResourceLoader resourceLoader) {
this.resourceLoader = resourceLoader;
}
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = applicationContext;
}
@Override
public Class<?extends FTPSendingMessageHandler> getObjectType() {
return FTPSendingMessageHandler.class;
}
@Override
protected FTPSendingMessageHandler createInstance()
throws Exception {
// the dependencies for the outbound-adapter are much simpler
// they only require an instance of the pool
DefaultFTPClientFactory defaultFTPClientFactory = new DefaultFTPClientFactory();
defaultFTPClientFactory.setHost(this.host);
defaultFTPClientFactory.setPassword(this.password);
defaultFTPClientFactory.setPort(this.port);
defaultFTPClientFactory.setRemoteWorkingDirectory(this.remoteDirectory);
defaultFTPClientFactory.setUsername(this.username);
defaultFTPClientFactory.setClientMode(this.clientMode);
QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool(15, defaultFTPClientFactory);
FTPSendingMessageHandler ftpSendingMessageHandler = new FTPSendingMessageHandler(queuedFTPClientPool);
ftpSendingMessageHandler.afterPropertiesSet();
return ftpSendingMessageHandler;
}
public void setPort(int port) {
this.port = port;
}
public void setUsername(String username) {
this.username = username;
}
public void setPassword(String password) {
this.password = password;
}
public void setHost(String host) {
this.host = host;
}
public void setRemoteDirectory(String remoteDirectory) {
this.remoteDirectory = remoteDirectory;
}
}

116
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java Normal file
View File

@@ -0,0 +1,116 @@
/*
* 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.ftp;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.net.ftp.FTPClient;
import org.springframework.util.Assert;
import java.io.IOException;
import java.net.SocketException;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
/**
* FTPClientPool implementation based on a Queue. This implementation has a
* default pool size of 5, but this is configurable with a constructor argument.
* <p>
* 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 {
private static final Log log = LogFactory.getLog(QueuedFTPClientPool.class);
private static final int DEFAULT_POOL_SIZE = 5;
private final Queue<FTPClient> pool;
private final FTPClientFactory factory;
public QueuedFTPClientPool(FTPClientFactory factory) {
this(DEFAULT_POOL_SIZE, factory);
}
/**
* @param maxPoolSize the maximum size of the pool
*/
public QueuedFTPClientPool(int maxPoolSize, FTPClientFactory factory) {
Assert.notNull(factory);
this.factory = factory;
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.
* <p>
* 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();
}
return prepareClient(client);
}
/**
* Prepares the client before it is returned through
* <code>getClient()</code>. The default implementation will check the
* connection using a noOp and replace the client with a new one if it
* encounters a problem.
* <p/>
* In more exotic environments subclasses can override this method to
* implement their own preparation strategy.
* @param client the unprepared client
* @return
* @throws SocketException
* @throws IOException
*/
protected FTPClient prepareClient(FTPClient client)
throws SocketException, IOException {
return isClientAlive(client) ? client : getClient();
}
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) {
if ((client != null) && !pool.offer(client)) {
try {
client.disconnect();
} catch (IOException e) {
log.warn("Error disconnecting ftpclient", e);
}
}
}
}

95
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FTPNamespaceHandler.java Normal file
View File

@@ -0,0 +1,95 @@
/*
* Copyright 2010 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.ftp.config;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.w3c.dom.Element;
import java.util.HashMap;
import java.util.Map;
/**
* Provides namespace support for using FTP
*
* @author Josh Long (*heavily* influenced by the good done by iwein before)
*/
@SuppressWarnings("unused")
public class FTPNamespaceHandler extends NamespaceHandlerSupport {
private static final String PACKAGE_NAME = "org.springframework.integration.ftp";
static private Map<String, Integer> CLIENT_MODES = new HashMap<String, Integer>();
static {
CLIENT_MODES.put("active-local-data-connection-mode", 0);
CLIENT_MODES.put("active-remote-data-connection-mode", 1);
CLIENT_MODES.put("passive-local-data-connection-mode", 2);
CLIENT_MODES.put("passive-remote-data-connection-mode", 3);
}
public void init() {
registerBeanDefinitionParser("inbound-channel-adapter", new FTPMessageSourceBeanDefinitionParser());
registerBeanDefinitionParser("outbound-channel-adapter", new FTPMessageSendingConsumerBeanDefinitionParser());
}
/**
* Configures an object that can take inbound messages and send them.
*/
private static class FTPMessageSendingConsumerBeanDefinitionParser extends AbstractOutboundChannelAdapterParser {
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PACKAGE_NAME + ".FTPSendingMessageHandlerFactoryBean");
for (String p : "auto-create-directories,username,port,password,host,key-file,key-file-password,remote-directory".split(",")) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, p);
}
int clientMode = CLIENT_MODES.get(element.getAttribute("client-mode"));
builder.addPropertyValue("clientMode", clientMode);
return builder.getBeanDefinition();
}
}
/**
* Configures an object that can recieve files from a remote SFTP endpoint and broadcast their arrival to the
* consumer
*/
private static class FTPMessageSourceBeanDefinitionParser extends AbstractPollingInboundChannelAdapterParser {
@Override
@SuppressWarnings("unused")
protected String parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PACKAGE_NAME + ".FTPMessageSourceFactoryBean");
for (String p : ("auto-create-directories,username,password,host,port," + "remote-directory,local-working-directory").split(",")) { //auto-delete-remote-files-on-sync
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, p);
}
int clientMode = CLIENT_MODES.get(element.getAttribute("client-mode"));
builder.addPropertyValue("clientMode", clientMode);
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
}
}
}

0
spring-integration-ftp/src/main/resources/META-INF/MANIFEST.MF Normal file
View File

1
spring-integration-ftp/src/main/resources/META-INF/spring.handlers Normal file
View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration/ftp=org.springframework.integration.ftp.config.FTPNamespaceHandler

2
spring-integration-ftp/src/main/resources/META-INF/spring.schemas Normal file
View File

@@ -0,0 +1,2 @@
http\://www.springframework.org/schema/integration/ftp/spring-integration-ftp-2.0.xsd=org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd
http\://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd=org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd

205
spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd Normal file
View File

@@ -0,0 +1,205 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2010 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.
-->
<xsd:schema xmlns="http://www.springframework.org/schema/integration/ftp"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/ftp"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.0.xsd"/>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation><![CDATA[
Builds an outbound-channel-adapter that writes files to a remote FTP endpoint.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="channel" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="username" type="xsd:string" use="required"/>
<xsd:attribute name="remote-directory" type="xsd:string" use="required"/>
<xsd:attribute name="host" type="xsd:string" use="required"/>
<xsd:attribute name="password" type="xsd:string"/>
<xsd:attribute name="port" type="xsd:int" default="22"/>
<xsd:attribute use="optional" name="client-mode" default="active-local-data-connection-mode">
<xsd:annotation>
<xsd:documentation><![CDATA[
the FTP Client-Mode.
/***
* A constant indicating the FTP session is expecting all transfers
* to occur between the client (local) and server and that the server
* should connect to the client's data port to initiate a data transfer.
* This is the default data connection mode when and FTPClient instance
* is created.
***/
ACTIVE_LOCAL_DATA_CONNECTION_MODE = 0
/***
* A constant indicating the FTP session is expecting all transfers
* to occur between two remote servers and that the server
* the client is connected to should connect to the other server's
* data port to initiate a data transfer.
***/
ACTIVE_LOCAL_DATA_CONNECTION_MODE = 1
/***
* A constant indicating the FTP session is expecting all transfers
* to occur between the client (local) and server and that the server
* is in passive mode, requiring the client to connect to the
* server's data port to initiate a transfer.
***/
PASSIVE_LOCAL_DATA_CONNECTION_MODE = 2
/***
* A constant indicating the FTP session is expecting all transfers
* to occur between two remote servers and that the server
* the client is connected to is in passive mode, requiring the other
* server to connect to the first server's data port to initiate a data
* transfer.
***/
PASSIVE_REMOTE_DATA_CONNECTION_MODE = 3
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="active-local-data-connection-mode"/>
<xsd:enumeration value="active-remote-data-connection-mode"/>
<xsd:enumeration value="passive-local-data-connection-mode"/>
<xsd:enumeration value="passive-remote-data-connection-mode"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation><![CDATA[
Builds an inbound-channel-adapter that synchronizes a local directory with the contents of a remote FTP endpoint.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="channel" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="local-working-directory" type="xsd:string"/>
<xsd:attribute name="auto-create-directories" type="xsd:boolean"/>
<xsd:attribute name="auto-delete-remote-files-on-sync" type="xsd:boolean"/>
<xsd:attribute name="username" type="xsd:string" use="required"/>
<xsd:attribute name="remote-directory" type="xsd:string" use="required"/>
<xsd:attribute name="host" type="xsd:string" use="required"/>
<xsd:attribute name="password" type="xsd:string"/>
<xsd:attribute name="port" type="xsd:int" default="22"/>
<xsd:attribute use="optional" name="client-mode" default="active-local-data-connection-mode">
<xsd:annotation>
<xsd:documentation><![CDATA[
the FTP Client-Mode.
/***
* A constant indicating the FTP session is expecting all transfers
* to occur between the client (local) and server and that the server
* should connect to the client's data port to initiate a data transfer.
* This is the default data connection mode when and FTPClient instance
* is created.
***/
ACTIVE_LOCAL_DATA_CONNECTION_MODE = 0
/***
* A constant indicating the FTP session is expecting all transfers
* to occur between two remote servers and that the server
* the client is connected to should connect to the other server's
* data port to initiate a data transfer.
***/
ACTIVE_LOCAL_DATA_CONNECTION_MODE = 1
/***
* A constant indicating the FTP session is expecting all transfers
* to occur between the client (local) and server and that the server
* is in passive mode, requiring the client to connect to the
* server's data port to initiate a transfer.
***/
PASSIVE_LOCAL_DATA_CONNECTION_MODE = 2
/***
* A constant indicating the FTP session is expecting all transfers
* to occur between two remote servers and that the server
* the client is connected to is in passive mode, requiring the other
* server to connect to the first server's data port to initiate a data
* transfer.
***/
PASSIVE_REMOTE_DATA_CONNECTION_MODE = 3
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="active-local-data-connection-mode"/>
<xsd:enumeration value="active-remote-data-connection-mode"/>
<xsd:enumeration value="passive-local-data-connection-mode"/>
<xsd:enumeration value="passive-remote-data-connection-mode"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:schema>

37
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/InboundFTPFileServiceActivator.java Normal file
View File

@@ -0,0 +1,37 @@
package org.springframework.integration.ftp;
import org.apache.commons.lang.StringUtils;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.stereotype.Component;
import java.io.File;
/**
* Simple component to test the inbound integration
*
* @author Josh Long
*/
public class InboundFTPFileServiceActivator {
@ServiceActivator
public void onNewRemoteFTPFile(Message<File> file)
throws Throwable {
System.out.println(StringUtils.repeat("=", 100));
System.out.println("A new file has appeared: " + file.getPayload().getAbsolutePath());
for (String h : file.getHeaders().keySet())
System.out.println(String.format("%s = %s", h, file.getHeaders().get(h)));
}
public static void main(String[] args) throws Throwable {
ClassPathXmlApplicationContext classPathXmlApplicationContext =
new ClassPathXmlApplicationContext("inbound-ftp-context.xml");
}
}

16
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/OutboundFTPExample.java Normal file
View File

@@ -0,0 +1,16 @@
package org.springframework.integration.ftp;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* This simple example demonstrates sending a file to a remote FTP server using the ftp:outbound-channel-adapter
*
* It reads files from a directory on your computer and systematically puts them on the remote FTP server,
*
* @author Josh Long
*/
public class OutboundFTPExample {
public static void main(String [] args ) throws Throwable {
ClassPathXmlApplicationContext classPathXmlApplicationContext = new ClassPathXmlApplicationContext("outbound-ftp-context.xml");
}
}

33
spring-integration-ftp/src/test/resources/inbound-ftp-context.xml Normal file
View File

@@ -0,0 +1,33 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:ftp="http://www.springframework.org/schema/integration/ftp"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:property-placeholder
location="file://${user.home}/Desktop/ftp.properties"
ignore-unresolvable="true"/>
<ftp:inbound-channel-adapter remote-directory="${ftp.remotedir}" channel="ftpIn" auto-create-directories="true"
host="${ftp.host}" auto-delete-remote-files-on-sync="false"
username="${ftp.username}" password="${ftp.password}" port="2222"
client-mode="passive-local-data-connection-mode"
>
<int:poller>
<int:interval-trigger interval="10000" time-unit="MILLISECONDS"/>
</int:poller>
</ftp:inbound-channel-adapter>
<int:channel id="ftpIn"/>
<bean id="inboundFTPFileServiceActivator"
class="org.springframework.integration.ftp.InboundFTPFileServiceActivator"/>
<int:service-activator input-channel="ftpIn" ref="inboundFTPFileServiceActivator"/>
</beans>

36
spring-integration-ftp/src/test/resources/outbound-ftp-context.xml Normal file
View File

@@ -0,0 +1,36 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:ftp="http://www.springframework.org/schema/integration/ftp"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:file="http://www.springframework.org/schema/integration/file"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file-1.0.xsd">
<context:property-placeholder
location="file://${user.home}/Desktop/ftp.properties"
ignore-unresolvable="true"/>
<file:inbound-channel-adapter channel="ftpOutbound"
filename-pattern=".*?jpg"
directory="#{systemProperties['user.home']}/Desktop/imagesToSendViaFTP"
auto-create-directory="true">
<int:poller>
<int:interval-trigger interval="1000" time-unit="MILLISECONDS"/>
</int:poller>
</file:inbound-channel-adapter>
<int:channel id="ftpOutbound"/>
<ftp:outbound-channel-adapter
remote-directory="${ftp.remotedir}" channel="ftpOutbound"
host="${ftp.host}"
username="${ftp.username}" password="${ftp.password}" port="2222"
client-mode="passive-local-data-connection-mode"
/>
</beans>

15
spring-integration-ftp/template.mf Normal file
View File

@@ -0,0 +1,15 @@
Bundle-SymbolicName: org.springframework.integration.ftp
Bundle-Name: Spring Integration SFTP Support
Bundle-Vendor: SpringSource
Bundle-ManifestVersion: 2
Import-Template:
org.apache.commons.logging;version="[1.1.1, 2.0.0)",
org.apache.commons.lang.*;version="[2.5.0, 3.0.0)",
org.springframework.integration.*;version="[2.0.0, 2.0.1)",
org.springframework.beans.*;version="[3.0.0, 4.0.0)",
org.springframework.context;version="[3.0.0, 4.0.0)",
org.springframework.core.*;version="[3.0.0, 4.0.0)",
org.springframework.util;version="[3.0.0, 4.0.0)",
org.jivesoftware.*;version="[3.1.0, 4.0.0)",
javax.*;version="0",
org.w3c.dom.*;version="0"

99
spring-integration-sftp/pom.xml Normal file
View File

@@ -0,0 +1,99 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-parent</artifactId>
<version>2.0.0.BUILD-SNAPSHOT</version>
</parent>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-sftp</artifactId>
<packaging>jar</packaging>
<name>Spring Integration SFTP Support</name>
<dependencies>
<dependency>
<groupId>javax.activation</groupId>
<artifactId>activation</artifactId>
<version>1.1.1</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-file</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.jcraft</groupId>
<artifactId>jsch</artifactId>
<version>0.1.42</version>
</dependency>
<dependency>
<groupId>cglib</groupId>
<artifactId>cglib-nodep</artifactId>
<version>${cglib.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.easymock</groupId>
<artifactId>easymock</artifactId>
<version>${org.easymock.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.easymock</groupId>
<artifactId>easymockclassextension</artifactId>
<version>${org.easymock.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>${junit.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context-support</artifactId>
<version>${org.springframework.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${org.springframework.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
<version>${project.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>${project.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.5</version>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>1.4</version>
</dependency>
</dependencies>
<build>
<!-- <plugins>
<plugin>
<groupId>com.springsource.bundlor</groupId>
<artifactId>com.springsource.bundlor.maven</artifactId>
</plugin>
</plugins> -->
</build>
</project>

98
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/QueuedSFTPSessionPool.java Normal file
View File

@@ -0,0 +1,98 @@
/*
* Copyright 2010 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.sftp;
import org.springframework.beans.factory.InitializingBean;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
/**
* This approach - of having a SessionPool ({@link org.springframework.integration.sftp.SFTPSessionPool}) that has an
* implementation of Queued*SessionPool ({@link org.springframework.integration.sftp.QueuedSFTPSessionPool}) - was
* taken pretty directly from the incredibly good Spring IntegrationFTP adapter.
*
* @author Josh Long
* @since 2.0
*/
public class QueuedSFTPSessionPool implements SFTPSessionPool, InitializingBean {
public static final int DEFAULT_POOL_SIZE = 10;
private Queue<SFTPSession> queue;
private final SFTPSessionFactory sftpSessionFactory;
private int maxPoolSize;
public QueuedSFTPSessionPool(SFTPSessionFactory factory) {
this(DEFAULT_POOL_SIZE, factory);
}
public QueuedSFTPSessionPool(int maxPoolSize, SFTPSessionFactory sessionFactory) {
this.sftpSessionFactory = sessionFactory;
this.maxPoolSize = maxPoolSize;
}
public void afterPropertiesSet() throws Exception {
assert maxPoolSize > 0 : "poolSize must be greater than 0!";
queue = new ArrayBlockingQueue<SFTPSession>(maxPoolSize, true); // size, faireness to avoid starvation
assert sftpSessionFactory != null : "sftpSessionFactory must not be null!";
}
public SFTPSession getSession() throws Exception {
SFTPSession session = this.queue.poll();
if (null == session) {
session = this.sftpSessionFactory.getObject();
if (queue.size() < maxPoolSize) {
queue.add(session);
}
}
if (null == session) {
session = queue.poll();
}
return session;
}
public void release(SFTPSession session) {
if (queue.size() < maxPoolSize) {
queue.add(session); // somehow one snuck in before <code>session</code> was finished!
} else {
dispose(session);
}
}
private void dispose(SFTPSession s) {
if (s == null) {
return;
}
if (queue.contains(s)) //this should never happen, but if it does ...
{
queue.remove(s);
}
if ((s.getChannel() != null) && s.getChannel().isConnected()) {
s.getChannel().disconnect();
}
if (s.getSession().isConnected()) {
s.getSession().disconnect();
}
}
}

25
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPConstants.java Normal file
View File

@@ -0,0 +1,25 @@
/*
* Copyright 2010 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.sftp;
/**
* @author Josh Long
*/
public class SFTPConstants {
public static final String SFTP_REMOTE_DIRECTORY_HEADER = "SFTP_REMOTE_DIRECTORY_HEADER";
}

277
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPInboundSynchronizer.java Normal file
View File

@@ -0,0 +1,277 @@
/*
* Copyright 2010 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.sftp;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.SftpATTRS;
import org.apache.commons.io.IOUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.Resource;
import org.springframework.integration.MessagingException;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Collection;
import java.util.concurrent.ScheduledFuture;
/**
*
* This handles keeping the {@link #localDirectory} in sync with the contents of the remote mount. From there, files are deposited
* into a folder where the {@link org.springframework.integration.file.FileReadingMessageSource} will eventually deliver them as events
*
*
* @author Josh Long
* @author Mario Gray
*/
public class SFTPInboundSynchronizer implements InitializingBean {
private static final long DEFAULT_REFRESH_RATE = 10 * 1000; // 10 seconds
// a lot of the approach for this (including the use of a FileReadingMessageSource and the regex / mask approach were lifted from FtpInboundSynchronizer
static final String INCOMPLETE_EXTENSION = ".INCOMPLETE";
private volatile Resource localDirectory;
private volatile SFTPSessionPool pool;
private volatile ScheduledFuture<?> scheduledFuture;
private volatile String remotePath;
private volatile TaskScheduler taskScheduler;
private volatile Trigger trigger = new PeriodicTrigger(DEFAULT_REFRESH_RATE);
private volatile boolean autoCreatePath;
private volatile boolean running;
private volatile boolean shouldDeleteDownloadedRemoteFiles; //.. this is false
public void afterPropertiesSet() throws Exception {
assert (taskScheduler != null) : "taskScheduler can't be null!";
assert (localDirectory != null) : "the localDirectory property must not be null!";
File localDir = localDirectory.getFile();
if (!localDir.exists()) {
if (autoCreatePath) {
if (!localDir.mkdirs()) {
throw new RuntimeException(String.format("couldn't create localDirectory %s", this.localDirectory.getFile().getAbsolutePath()));
}
}
}
}
public ScheduledFuture<?> getScheduledFuture() {
return scheduledFuture;
}
public TaskScheduler getTaskScheduler() {
return taskScheduler;
}
public Trigger getTrigger() {
return trigger;
}
public boolean isRunning() {
return running;
}
public boolean isShouldDeleteDownloadedRemoteFiles() {
return shouldDeleteDownloadedRemoteFiles;
}
public void setAutoCreatePath(boolean autoCreatePath) {
this.autoCreatePath = autoCreatePath;
}
public void setLocalDirectory(Resource localDirectory) {
this.localDirectory = localDirectory;
}
public void setPool(SFTPSessionPool pool) {
this.pool = pool;
}
public void setRemotePath(String remotePath) {
this.remotePath = remotePath;
}
public void setScheduledFuture(ScheduledFuture<?> scheduledFuture) {
this.scheduledFuture = scheduledFuture;
}
public void setShouldDeleteDownloadedRemoteFiles(boolean shouldDeleteDownloadedRemoteFiles) {
this.shouldDeleteDownloadedRemoteFiles = shouldDeleteDownloadedRemoteFiles;
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
}
public void setTrigger(Trigger t) {
this.trigger = t;
}
public void start() {
if (running) {
return;
}
assert checkThatRemotePathExists(remotePath) : "the remotePath had better exist!";
assert taskScheduler != null : "'taskScheduler' is required";
scheduledFuture = taskScheduler.schedule(new SynchronizeTask(), trigger);
this.running = true;
}
public void stop() {
if (!running) {
return;
}
assert scheduledFuture != null : "scheduledFuture is null!";
this.scheduledFuture.cancel(true);
this.running = false;
}
@SuppressWarnings("unchecked")
public void synchronize() throws Exception {
SFTPSession session = null;
try {
session = pool.getSession();
session.start();
ChannelSftp channelSftp = session.getChannel();
Collection<ChannelSftp.LsEntry> files = channelSftp.ls(remotePath);
for (ChannelSftp.LsEntry lsEntry : files) {
if ((lsEntry != null) && !lsEntry.getAttrs().isDir() && !lsEntry.getAttrs().isLink()) {
copyFromRemoteToLocalDirectory(session, lsEntry, this.localDirectory);
}
}
} catch (IOException e) {
throw new MessagingException("couldn't synchronize remote to local directory", e);
} finally {
if ((session != null) && (pool != null)) {
pool.release(session);
}
}
}
/**
* there be dragons this way ... This method will check to ensure that the remote directory exists. If the directory
* doesnt exist, and autoCreatePath is configured to be true, then this method makes a few reasonably sane attempts
* to create it. Otherwise, it fails fast.
*
* @param rPath the path on the remote SSH / SFTP server to create.
* @return whether or not the directory is there (regardless of whether we created it in this method or it already
* existed.)
*/
private boolean checkThatRemotePathExists(String rPath) {
SFTPSession session = null;
ChannelSftp channelSftp = null;
try {
session = pool.getSession();
assert session != null : "session's not null";
session.start();
channelSftp = session.getChannel();
SftpATTRS attrs = channelSftp.stat(rPath);
assert (attrs != null) && attrs.isDir() : "attrs can't be null, and should indicate that it's a directory!";
return true;
} catch (Throwable th) {
if (this.autoCreatePath && (pool != null) && (session != null)) {
try {
if (channelSftp != null) {
channelSftp.mkdir(rPath);
if (channelSftp.stat(rPath).isDir()) {
return true;
}
}
} catch (Throwable t) {
return false;
}
}
} finally {
if ((pool != null) && (session != null)) {
pool.release(session);
}
}
return false;
}
@SuppressWarnings("ignored")
private boolean copyFromRemoteToLocalDirectory(SFTPSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir)
throws Exception {
File fileForLocalDir = localDir.getFile();
File localFile = new File(fileForLocalDir, entry.getFilename());
if (!localFile.exists()) {
InputStream in = null;
FileOutputStream fos = null;
try {
File tmpLocalTarget = new File(localFile.getAbsolutePath() + INCOMPLETE_EXTENSION);
fos = new FileOutputStream(tmpLocalTarget);
String remoteFqPath = this.remotePath + "/" + entry.getFilename();
in = sftpSession.getChannel().get(remoteFqPath);
IOUtils.copy(in, fos);
if (tmpLocalTarget.renameTo(localFile)) {
// last step
if (isShouldDeleteDownloadedRemoteFiles()) {
sftpSession.getChannel().rm(remoteFqPath);
}
}
return true;
} catch (Throwable th) {
IOUtils.closeQuietly(in);
IOUtils.closeQuietly(fos);
}
} else {
return true;
}
return false;
}
private boolean foo() {
return false;
}
class SynchronizeTask implements Runnable {
public void run() {
try {
synchronize();
} catch (Throwable e) {
// todo logger.debug("couldn't invoke synchronize()", e);
}
}
}
}

131
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPMessageSource.java Normal file
View File

@@ -0,0 +1,131 @@
/*
* Copyright 2010 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.sftp;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.io.Resource;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.AcceptOnceFileListFilter;
import org.springframework.integration.file.CompositeFileListFilter;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.PatternMatchingFileListFilter;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import java.io.File;
import java.io.IOException;
import java.util.regex.Pattern;
/**
* this creates the message source that ultimately 'see's files on a local directory and forwards them on to the bus.
* These files are asynchronously deposited into a folder via the SFTP synchronizer. This code is <i>very</i> influenced
* by the FtpFileSource class from the Spring Integration FTP adapter.
*
* @author Josh Long
*/
public class SFTPMessageSource implements MessageSource<File>, InitializingBean, Lifecycle {
private FileReadingMessageSource fileReadingMessageSource;
private Resource localDirectory;
private SFTPInboundSynchronizer synchronizer;
private TaskScheduler taskScheduler;
private Trigger trigger;
public SFTPMessageSource(FileReadingMessageSource fileSource, SFTPInboundSynchronizer synchronizer) {
this.fileReadingMessageSource = fileSource;
this.synchronizer = synchronizer;
Pattern completePattern = Pattern.compile("^.*(?<!" + SFTPInboundSynchronizer.INCOMPLETE_EXTENSION + ")$");
fileReadingMessageSource.setFilter(new CompositeFileListFilter(new AcceptOnceFileListFilter(), new PatternMatchingFileListFilter(completePattern)));
}
public void afterPropertiesSet() throws Exception {
synchronizer.afterPropertiesSet();
this.fileReadingMessageSource.afterPropertiesSet();
}
public FileReadingMessageSource getFileReadingMessageSource() {
return fileReadingMessageSource;
}
public Resource getLocalDirectory() {
return localDirectory;
}
public SFTPInboundSynchronizer getSynchronizer() {
return synchronizer;
}
public TaskScheduler getTaskScheduler() {
return taskScheduler;
}
public Trigger getTrigger() {
return trigger;
}
public boolean isRunning() {
return this.synchronizer.isRunning();
}
public Message<File> receive() {
return this.fileReadingMessageSource.receive();
}
public void setFileReadingMessageSource(final FileReadingMessageSource fileReadingMessageSource) {
this.fileReadingMessageSource = fileReadingMessageSource;
}
public void setLocalDirectory(final Resource localDirectory) {
this.localDirectory = localDirectory;
try {
this.fileReadingMessageSource.setDirectory(localDirectory.getFile());
} catch (IOException e) {
}
this.synchronizer.setLocalDirectory(localDirectory);
}
public void setSynchronizer(final SFTPInboundSynchronizer synchronizer) {
this.synchronizer = synchronizer;
}
public void setTaskScheduler(final TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
synchronizer.setTaskScheduler(taskScheduler);
}
public void setTrigger(final Trigger trigger) {
this.trigger = trigger;
this.synchronizer.setTrigger(trigger);
}
public void start() {
synchronizer.start();
}
public void stop() {
synchronizer.stop();
}
}

140
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSendingMessageHandler.java Normal file
View File

@@ -0,0 +1,140 @@
/*
* Copyright 2010 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.sftp;
import com.jcraft.jsch.ChannelSftp;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.*;
import org.springframework.integration.core.MessageHandler;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
/**
* Sending a message payload to a remote SFTP endpoint. For now, we assume that the payload of the inbound message is of
* type {@link java.io.File}. Perhaps we could support a payload of java.io.InputStream with a Header designating the file
* name?
*
* @author Josh Long
*/
public class SFTPSendingMessageHandler implements MessageHandler, InitializingBean {
private SFTPSessionPool pool;
private String remoteDirectory;
private volatile boolean afterPropertiesSetRan;
public SFTPSendingMessageHandler(SFTPSessionPool pool) {
this.pool = pool;
}
public void afterPropertiesSet() throws Exception {
assert this.pool != null : "the pool can't be null!";
// logger.debug("afterPropertiesSet() called on SFTPSendingMessageHandler");
if (!afterPropertiesSetRan) {
if (StringUtils.isEmpty(this.remoteDirectory)) {
remoteDirectory = null;
}
this.afterPropertiesSetRan = true;
}
}
public String getRemoteDirectory() {
return remoteDirectory;
}
public void handleMessage(final Message<?> message)
throws MessageRejectedException, MessageHandlingException, MessageDeliveryException {
assert this.pool != null : "need a working pool";
assert message.getPayload() instanceof File : "the payload needs to be java.io.File";
try {
File inboundFilePayload = (File) message.getPayload();
if ((inboundFilePayload != null) && inboundFilePayload.exists()) {
sendFileToRemoteEndpoint(message, inboundFilePayload);
}
} catch (Throwable thr) {
// logger.debug("recieved an exception.", thr);
throw new MessageDeliveryException(message, "couldn't deliver the message!", thr);
}
}
public void setRemoteDirectory(final String remoteDirectory) {
this.remoteDirectory = remoteDirectory;
}
private boolean sendFileToRemoteEndpoint(Message<?> message, File file)
throws Throwable {
assert this.pool != null : "need a working pool";
SFTPSession session = this.pool.getSession();
if (session == null) {
throw new RuntimeException("the session returned from the pool is null, can't possibly proceed.");
}
session.start();
ChannelSftp sftp = session.getChannel();
InputStream fileInputStream = null;
try {
fileInputStream = new FileInputStream(file);
String baseOfRemotePath = StringUtils.isEmpty(this.remoteDirectory) ? StringUtils.EMPTY : remoteDirectory; // the safe default
// logger.debug("going to send " + file.getAbsolutePath() + " to a remote sftp endpoint");
String dynRd = null;
MessageHeaders messageHeaders = null;
if (message != null) {
messageHeaders = message.getHeaders();
if ((messageHeaders != null) && messageHeaders.containsKey(SFTPConstants.SFTP_REMOTE_DIRECTORY_HEADER)) {
dynRd = (String) messageHeaders.get(SFTPConstants.SFTP_REMOTE_DIRECTORY_HEADER);
if (!StringUtils.isEmpty(dynRd)) {
baseOfRemotePath = dynRd;
}
}
}
if (!StringUtils.defaultString(baseOfRemotePath).endsWith("/")) {
baseOfRemotePath += "/";
}
sftp.put(fileInputStream, baseOfRemotePath + file.getName());
return true;
} finally {
IOUtils.closeQuietly(fileInputStream);
if (pool != null) {
pool.release(session);
}
}
}
}

159
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSession.java Normal file
View File

@@ -0,0 +1,159 @@
/*
* Copyright 2010 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.sftp;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.Session;
import com.jcraft.jsch.UserInfo;
import org.apache.commons.lang.StringUtils;
import java.io.InputStream;
/**
* There are many ways to create a {@link org.springframework.integration.sftp.SFTPSession} just as there are many ways to SSH into a remote system.
* You may use a username and password, you may use a username and private key, you may use a username and a private key with a password, etc.
*
* This object represents the connection to the remote server, and to use it you must provide it with all the components you'd normally provide an
* incantation of the <code>ssh</code> command.
*
*
* @author Josh Long
* @author Mario Gray
*/
public class SFTPSession {
private volatile ChannelSftp channel;
private volatile Session session;
private String privateKey;
private String privateKeyPassphrase;
private volatile UserInfo userInfo;
/**
* @param userName the name of the account being logged into.
* @param hostName this should be the host. I found values like <code>foo.com</code> work, where
* <code>http://foo.com</code> don't.
* @param userPassword if you are not using key based authentication, then you are likely being prompted
* for a password each time you login. This is that password. It is <em>not</em> the
* passphrase for the private key!
* @param port the default is 22, and if you specify N<0 for this value we'll default it to 22
* @param knownHostsFile this is the known hosts file. If you don't specify it, jsch does some magic to work
* without your specification. If you have it in a non well-known location, however,
* this property is for you. An example: <code>/home/user/.ssh/known_hosts</code>
* @param knownHostsInputStream this is the known hosts file. If you don't specify it, jsch does some magic to work
* without your specification. If you have it in a non well-known location, however,
* this property is for you. An example: <code>/home/user/.ssh/known_hosts</code>. Note
* that you may specify this <em>or</em> the #knownHostsFile - not both!
* @param privateKey this is usually used when you want passwordless automation (obviously, for this
* integration it's useless since this lets you specify a password once, anyway, but
* still good to have if required). This file might be ~/.ssh/id_dsa, or a
* <code>.pem</code> for your remote server (for example, on EC2)
* @param pvKeyPassPhrase sometimes, to be extra secure, the private key itself is extra encrypted. In order
* to surmount that, we need the private key passphrase. Specify that here.
*
* @throws Exception thrown if any of a myriad of scenarios plays out
*/
public SFTPSession(String userName, String hostName, String userPassword, int port, String knownHostsFile, InputStream knownHostsInputStream, String privateKey, String pvKeyPassPhrase)
throws Exception {
JSch jSch = new JSch();
if (port <= 0) {
port = 22;
}
this.privateKey = privateKey;
this.privateKeyPassphrase = pvKeyPassPhrase;
if (!StringUtils.isEmpty(knownHostsFile)) {
jSch.setKnownHosts(knownHostsFile);
} else if (null != knownHostsInputStream) {
jSch.setKnownHosts(knownHostsInputStream);
}
// private key
if (!StringUtils.isEmpty(this.privateKey)) {
if (!StringUtils.isEmpty(privateKeyPassphrase)) {
jSch.addIdentity(this.privateKey, privateKeyPassphrase);
} else {
jSch.addIdentity(this.privateKey);
}
}
session = jSch.getSession(userName, hostName, port);
if (!StringUtils.isEmpty(userPassword)) {
session.setPassword(userPassword);
}
userInfo = new OptimisticUserInfoImpl(userPassword);
session.setUserInfo(userInfo);
session.connect();
channel = (ChannelSftp) session.openChannel("sftp");
}
public ChannelSftp getChannel() {
return channel;
}
public Session getSession() {
return session;
}
public void start() throws Exception {
if (!channel.isConnected()) {
channel.connect();
}
}
/**
* this is a simple, optimistic implementation of this interface. It simply returns in the positive where possible
* and handles interactive authentication (ie, 'Please enter your password: ' prompts are dispatched automatically using this)
*/
private static class OptimisticUserInfoImpl implements UserInfo {
private String pw;
public OptimisticUserInfoImpl(String password) {
this.pw = password;
}
public String getPassphrase() {
return null; // pass
}
public String getPassword() {
return pw;
}
public boolean promptPassphrase(String string) {
return true;
}
public boolean promptPassword(String string) {
return true;
}
public boolean promptYesNo(String string) {
return true;
}
public void showMessage(String string) {
}
}
}

115
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSessionFactory.java Normal file
View File

@@ -0,0 +1,115 @@
/*
* Copyright 2010 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.sftp;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
/**
* Factories {@link org.springframework.integration.sftp.SFTPSession} instances. There are lots of ways to construct a
* {@link org.springframework.integration.sftp.SFTPSession} instance, and not all of them are obvious. This factory
* does its best to make it work.
*
* @author Josh Long
* @author Mario Gray
*/
public class SFTPSessionFactory implements FactoryBean<SFTPSession>, InitializingBean {
private volatile String knownHosts;
private volatile String password;
private volatile String privateKey;
private volatile String privateKeyPassphrase;
private volatile String remoteHost;
private volatile String user;
private volatile int port = 22; // the default
public void afterPropertiesSet() throws Exception {
assert !StringUtils.isEmpty(this.remoteHost) : "remoteHost can't be empty!";
assert !StringUtils.isEmpty(this.user) : "user can't be empty!";
assert !StringUtils.isEmpty(this.password) || !StringUtils.isEmpty(this.privateKey) || !StringUtils.isEmpty(this.privateKeyPassphrase) : "you must configure either a password or a private key and/or a private key passphrase!";
assert this.port >= 0 : "port must be a valid number! ";
}
public String getKnownHosts() {
return knownHosts;
}
public SFTPSession getObject() throws Exception {
return new SFTPSession(this.getUser(), this.getRemoteHost(), this.getPassword(), this.getPort(), this.getKnownHosts(), null, this.getPrivateKey(), this.getPrivateKeyPassphrase());
}
public Class<?extends SFTPSession> getObjectType() {
return SFTPSession.class;
}
public String getPassword() {
return password;
}
public int getPort() {
return port;
}
public String getPrivateKey() {
return privateKey;
}
public String getPrivateKeyPassphrase() {
return privateKeyPassphrase;
}
public String getRemoteHost() {
return remoteHost;
}
public String getUser() {
return user;
}
public boolean isSingleton() {
return false;
}
public void setKnownHosts(String knownHosts) {
this.knownHosts = knownHosts;
}
public void setPassword(String password) {
this.password = password;
}
public void setPort(int port) {
this.port = port;
}
public void setPrivateKey(String privateKey) {
this.privateKey = privateKey;
}
public void setPrivateKeyPassphrase(String privateKeyPassphrase) {
this.privateKeyPassphrase = privateKeyPassphrase;
}
public void setRemoteHost(String remoteHost) {
this.remoteHost = remoteHost;
}
public void setUser(String user) {
this.user = user;
}
}

44
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSessionPool.java Normal file
View File

@@ -0,0 +1,44 @@
/*
* Copyright 2010 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.sftp;
/**
*
* Holds instances of {@link org.springframework.integration.sftp.SFTPSession} since they're stateful
* and might be in use while another run happens.
*
*
* @author Josh Long
*/
public interface SFTPSessionPool {
/**
* this returns a session that can be used to connct to an sftp instance and perform operations
*
* @return the session from the pool ready to be connected to.
* @throws Exception thrown if theres any of the numerous faults possible when trying to connect to the remote
* server
*/
SFTPSession getSession() throws Exception;
/**
* Frees up the client. Im not sure what the meaningful semantics of this are. Perhaps it just calls <code>(session
* ,channel).disconnect()</code> ?
*
* @param session the session to relinquish / renew
*/
void release(SFTPSession session);
}

133
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSendingConsumerFactoryBean.java Normal file
View File

@@ -0,0 +1,133 @@
/*
* Copyright 2010 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.sftp.config;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.sftp.QueuedSFTPSessionPool;
import org.springframework.integration.sftp.SFTPSendingMessageHandler;
import org.springframework.integration.sftp.SFTPSessionFactory;
/**
* Supports the construction of a MessagHandler that knows how to take inbound #java.io.File objects and send them to a
* remote destination.
*
* @author Josh Long
*/
public class SFTPMessageSendingConsumerFactoryBean implements InitializingBean, FactoryBean<SFTPSendingMessageHandler> {
private String host;
private String keyFile;
private String keyFilePassword;
private String password;
private String remoteDirectory;
private String username;
private boolean autoCreateDirectories;
private int port;
public void afterPropertiesSet() throws Exception {
if (isAutoCreateDirectories()) {
// todo figure out this value
}
}
public String getHost() {
return host;
}
public String getKeyFile() {
return keyFile;
}
public String getKeyFilePassword() {
return keyFilePassword;
}
public SFTPSendingMessageHandler getObject() throws Exception {
SFTPSessionFactory sessionFactory = SFTPSessionUtils.buildSftpSessionFactory(this.getHost(), this.getPassword(), this.getUsername(), this.getKeyFile(), this.getKeyFilePassword(),
this.getPort());
QueuedSFTPSessionPool queuedSFTPSessionPool = new QueuedSFTPSessionPool(15, sessionFactory);
queuedSFTPSessionPool.afterPropertiesSet();
SFTPSendingMessageHandler sftpSendingMessageHandler = new SFTPSendingMessageHandler(queuedSFTPSessionPool);
sftpSendingMessageHandler.setRemoteDirectory(this.getRemoteDirectory());
sftpSendingMessageHandler.afterPropertiesSet();
return sftpSendingMessageHandler;
}
public Class<?extends SFTPSendingMessageHandler> getObjectType() {
return SFTPSendingMessageHandler.class;
}
public String getPassword() {
return password;
}
public int getPort() {
return port;
}
public String getRemoteDirectory() {
return remoteDirectory;
}
public String getUsername() {
return username;
}
public boolean isAutoCreateDirectories() {
return autoCreateDirectories;
}
public boolean isSingleton() {
return false;
}
public void setAutoCreateDirectories(final boolean autoCreateDirectories) {
this.autoCreateDirectories = autoCreateDirectories;
}
public void setHost(final String host) {
this.host = host;
}
public void setKeyFile(final String keyFile) {
this.keyFile = keyFile;
}
public void setKeyFilePassword(final String keyFilePassword) {
this.keyFilePassword = keyFilePassword;
}
public void setPassword(final String password) {
this.password = password;
}
public void setPort(final int port) {
this.port = port;
}
public void setRemoteDirectory(final String remoteDirectory) {
this.remoteDirectory = remoteDirectory;
}
public void setUsername(final String username) {
this.username = username;
}
}

277
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSourceFactoryBean.java Normal file
View File

@@ -0,0 +1,277 @@
/*
* Copyright 2010 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.sftp.config;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.SystemUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceEditor;
import org.springframework.core.io.ResourceLoader;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.sftp.QueuedSFTPSessionPool;
import org.springframework.integration.sftp.SFTPInboundSynchronizer;
import org.springframework.integration.sftp.SFTPMessageSource;
import org.springframework.integration.sftp.SFTPSessionFactory;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ErrorHandler;
import java.io.File;
import java.util.Map;
/**
*
* Building a {@link org.springframework.integration.sftp.SFTPMessageSource} is a complicated because we also
* use a {@link org.springframework.integration.file.FileReadingMessageSource} to handle the "receipt" of files in
* a {@link #localWorkingDirectory}.
*
* @author Josh Long
*/
public class SFTPMessageSourceFactoryBean extends AbstractFactoryBean<SFTPMessageSource> implements ApplicationContextAware, ResourceLoaderAware {
private ApplicationContext applicationContext;
private FileReadingMessageSource fileReadingMessageSource;
private Resource localDirectoryResource;
private ResourceLoader resourceLoader;
private SFTPInboundSynchronizer synchronizer;
private String host;
private String keyFile;
private String keyFilePassword;
private String localWorkingDirectory;
private String password;
private String remoteDirectory;
private String username;
private TaskScheduler taskScheduler;
private Trigger trigger;
private boolean autoCreateDirectories;
private boolean autoDeleteRemoteFilesOnSync;
private int port = 22;
public FileReadingMessageSource getFileReadingMessageSource() {
return fileReadingMessageSource;
}
public String getHost() {
return host;
}
public String getKeyFile() {
return keyFile;
}
public String getKeyFilePassword() {
return keyFilePassword;
}
public String getLocalWorkingDirectory() {
return localWorkingDirectory;
}
@Override
public Class<?extends SFTPMessageSource> getObjectType() {
return SFTPMessageSource.class;
}
public String getPassword() {
return password;
}
public int getPort() {
return port;
}
public String getRemoteDirectory() {
return remoteDirectory;
}
public SFTPInboundSynchronizer getSynchronizer() {
return synchronizer;
}
public TaskScheduler getTaskScheduler() {
return taskScheduler;
}
public Trigger getTrigger() {
return trigger;
}
// this is the ultimate layer of control
// users will configure theeir entire experience using this class and trust that a working
// component comes out as a result of their input
// we need to support user/pw/keys/host/port/auto-delete properties
public String getUsername() {
return username;
}
public boolean isAutoCreateDirectories() {
return autoCreateDirectories;
}
public boolean isAutoDeleteRemoteFilesOnSync() {
return autoDeleteRemoteFilesOnSync;
}
public void setApplicationContext(final ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = applicationContext;
}
public void setAutoCreateDirectories(final boolean autoCreateDirectories) {
this.autoCreateDirectories = autoCreateDirectories;
}
public void setAutoDeleteRemoteFilesOnSync(final boolean autoDeleteRemoteFilesOnSync) {
this.autoDeleteRemoteFilesOnSync = autoDeleteRemoteFilesOnSync;
}
public void setFileReadingMessageSource(final FileReadingMessageSource fileReadingMessageSource) {
this.fileReadingMessageSource = fileReadingMessageSource;
}
public void setHost(final String host) {
this.host = host;
}
public void setKeyFile(final String keyFile) {
this.keyFile = keyFile;
}
public void setKeyFilePassword(final String keyFilePassword) {
this.keyFilePassword = keyFilePassword;
}
public void setLocalWorkingDirectory(final String lwd) {
this.localWorkingDirectory = lwd;
}
public void setPassword(final String password) {
this.password = password;
}
public void setPort(final int port) {
this.port = port;
}
public void setRemoteDirectory(final String remoteDirectory) {
this.remoteDirectory = remoteDirectory;
}
public void setResourceLoader(final ResourceLoader resourceLoader) {
this.resourceLoader = resourceLoader;
}
public void setSynchronizer(final SFTPInboundSynchronizer synchronizer) {
this.synchronizer = synchronizer;
}
public void setTaskScheduler(final TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
}
public void setTrigger(final Trigger trigger) {
this.trigger = trigger;
}
public void setUsername(final String username) {
this.username = username;
}
@Override
protected SFTPMessageSource createInstance() throws Exception {
try {
if ((localWorkingDirectory == null) || StringUtils.isEmpty(localWorkingDirectory)) {
File tmp = SystemUtils.getJavaIoTmpDir();
File sftpTmp = new File(tmp, "sftpInbound");
this.localWorkingDirectory = "file://" + sftpTmp.getAbsolutePath();
}
assert !StringUtils.isEmpty(this.localWorkingDirectory) : "the local working directory mustn't be null!";
// resource for local directory
ResourceEditor editor = new ResourceEditor(this.resourceLoader);
editor.setAsText(this.localWorkingDirectory);
this.localDirectoryResource = (Resource) editor.getValue();
fileReadingMessageSource = new FileReadingMessageSource();
synchronizer = new SFTPInboundSynchronizer();
if (null == taskScheduler) {
Map<String, TaskScheduler> tss = null;
if ((tss = applicationContext.getBeansOfType(TaskScheduler.class)).keySet().size() != 0) {
taskScheduler = tss.get(tss.keySet().iterator().next());
}
}
if (null == taskScheduler) {
ThreadPoolTaskScheduler ts = new ThreadPoolTaskScheduler();
ts.setPoolSize(10);
ts.setErrorHandler(new ErrorHandler() {
public void handleError(Throwable t) {
// todo make this forward a message onto the error channel (how does that work?)
logger.debug("error! ", t);
}
});
ts.setWaitForTasksToCompleteOnShutdown(true);
ts.initialize();
this.taskScheduler = ts;
}
SFTPSessionFactory sessionFactory = SFTPSessionUtils.buildSftpSessionFactory(this.getHost(), this.getPassword(), this.getUsername(), this.getKeyFile(), this.getKeyFilePassword(),
this.getPort());
QueuedSFTPSessionPool pool = new QueuedSFTPSessionPool(15, sessionFactory);
pool.afterPropertiesSet();
synchronizer.setRemotePath(this.getRemoteDirectory());
synchronizer.setPool(pool);
synchronizer.setAutoCreatePath(this.isAutoCreateDirectories());
synchronizer.setShouldDeleteDownloadedRemoteFiles(this.isAutoDeleteRemoteFilesOnSync());
SFTPMessageSource sftpMessageSource = new SFTPMessageSource(fileReadingMessageSource, synchronizer);
sftpMessageSource.setTaskScheduler(taskScheduler);
if (null != this.trigger) {
sftpMessageSource.setTrigger(trigger);
}
sftpMessageSource.setLocalDirectory(this.localDirectoryResource);
sftpMessageSource.afterPropertiesSet();
sftpMessageSource.start();
return sftpMessageSource;
} catch (Throwable thr) {
logger.debug("error occurred when trying to configure SFTPmessageSource ", thr);
}
return null;
}
}

79
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPNamespaceHandler.java Normal file
View File

@@ -0,0 +1,79 @@
/*
* Copyright 2010 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.sftp.config;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.w3c.dom.Element;
/**
*
* Provides namespace support for using SFTP
*
* @author Josh Long
*/
@SuppressWarnings("unused")
public class SFTPNamespaceHandler extends NamespaceHandlerSupport {
private static final String PACKAGE_NAME = "org.springframework.integration.sftp";
public void init() {
registerBeanDefinitionParser("inbound-channel-adapter", new SFTPMessageSourceBeanDefinitionParser());
registerBeanDefinitionParser("outbound-channel-adapter", new SFTPMessageSendingConsumerBeanDefinitionParser());
}
/**
* Configures an object that can take inbound messages and send them.
*/
private static class SFTPMessageSendingConsumerBeanDefinitionParser extends AbstractOutboundChannelAdapterParser {
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PACKAGE_NAME + ".config.SFTPMessageSendingConsumerFactoryBean");
for (String p : "auto-create-directories,username,password,host,key-file,key-file-password,remote-directory".split(",")) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, p);
}
return builder.getBeanDefinition();
}
}
/**
* Configures an object that can recieve files from a remote SFTP endpoint and broadcast their arrival to the
* consumer
*/
private static class SFTPMessageSourceBeanDefinitionParser extends AbstractPollingInboundChannelAdapterParser {
@Override
@SuppressWarnings("unused")
protected String parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PACKAGE_NAME + ".config.SFTPMessageSourceFactoryBean");
for (String p : "auto-create-directories,username,password,host,key-file,key-file-password,remote-directory,local-working-directory,auto-delete-remote-files-on-sync".split(",")) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, p);
}
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
}
}
}

58
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPSessionUtils.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2010 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.sftp.config;
import org.springframework.integration.sftp.SFTPSessionFactory;
/**
*
* Provides a single place to handle this tedious chore.
*
* todo : replace all the ad-hoc definitions of {@link org.springframework.integration.sftp.SFTPSessionFactory}
*
* @author Josh Long
*/
public class SFTPSessionUtils {
/**
* This method hides the minutae required to build an #SFTPSessionFactory.
*
* @param host the host to connect to.
* @param usr this is required. It is the username of the credentials being authenticated.
* @param pw if password authentication is being used (as opposed to key-based authentication) then this is
* where you configure the password.
* @param pvKey the file that is the private key
* @param pvKeyPass the passphrase used to use the key file
* @param port the default (22) is used if the value here is N< 0. The value should be only be set if the port
* is non-standard (not 22)
* @return the SFTPSessionFactory that's used to create connections and get us in the right state to start issue
* commands against a remote SFTP/SSH filesystem
* @throws Exception thrown in case of darned near <em>anything</em>
*/
public static SFTPSessionFactory buildSftpSessionFactory(String host, String pw, String usr, String pvKey, String pvKeyPass, int port)
throws Exception {
SFTPSessionFactory sftpSessionFactory = new SFTPSessionFactory();
sftpSessionFactory.setPassword(pw);
sftpSessionFactory.setPort(port);
sftpSessionFactory.setRemoteHost(host);
sftpSessionFactory.setUser(usr);
sftpSessionFactory.setPrivateKey(pvKey);
sftpSessionFactory.setPrivateKeyPassphrase(pvKeyPass);
sftpSessionFactory.afterPropertiesSet();
return sftpSessionFactory;
}
}

30
spring-integration-sftp/src/main/resources/JSCH_LICENSE.txt Normal file
View File

@@ -0,0 +1,30 @@
JSch 0.0.* was released under the GNU LGPL license. Later, we have switched
over to a BSD-style license.
------------------------------------------------------------------------------
Copyright (c) 2002,2003,2004,2005,2006,2007,2008 Atsuhiko Yamanaka, JCraft,Inc.
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
1. Redistributions of source code must retain the above copyright notice,
this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in
the documentation and/or other materials provided with the distribution.
3. The names of the authors may not be used to endorse or promote products
derived from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES,
INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL JCRAFT,
INC. OR ANY CONTRIBUTORS TO THIS SOFTWARE BE LIABLE FOR ANY DIRECT, INDIRECT,
INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

0
spring-integration-sftp/src/main/resources/META-INF/MANIFEST.MF Normal file
View File

1
spring-integration-sftp/src/main/resources/META-INF/spring.handlers Normal file
View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration/sftp=org.springframework.integration.sftp.config.SFTPNamespaceHandler

2
spring-integration-sftp/src/main/resources/META-INF/spring.schemas Normal file
View File

@@ -0,0 +1,2 @@
http\://www.springframework.org/schema/integration/sftp/spring-integration-sftp-2.0.xsd=org/springframework/integration/sftp/config/spring-integration-sftp-2.0.xsd
http\://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd=org/springframework/integration/sftp/config/spring-integration-sftp-2.0.xsd

117
spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.0.xsd Normal file
View File

@@ -0,0 +1,117 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2010 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.
-->
<xsd:schema xmlns="http://www.springframework.org/schema/integration/sftp"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/sftp"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.0.xsd"/>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation><![CDATA[
Builds an outbound-channel-adapter that writes files to a remote SFTP endpoint.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="channel" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="username" type="xsd:string" use="required"/>
<xsd:attribute name="remote-directory" type="xsd:string" use="required"/>
<xsd:attribute name="host" type="xsd:string" use="required"/>
<xsd:attribute name="password" type="xsd:string"/>
<xsd:attribute name="port" type="xsd:int" default="22"/>
<xsd:attribute name="key-file" type="xsd:string"/>
<xsd:attribute name="key-file-password" type="xsd:string"/>
<xsd:attribute name="auto-create-directories" type="xsd:boolean"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation><![CDATA[
Builds an inbound-channel-adapter that synchronizes with a remote SFTP endpoint.
You may configure a <code>poller</code> element to determine at what frequency to scan the remote directory.
There is support for automatically deleting remote files upon synchornization. This adapter supports two connectivity options:
<ol>
<li> Password authentication: using this opton, authenticatio is done using a username and a password.</li>
<li> Key-based authentication: using this option, you may specify a key that will be used to authenticate. If they key itself is encrypted and requires a password, you may specify that, as well.</li>
</ol>
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="channel" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="username" type="xsd:string" use="required"/>
<xsd:attribute name="remote-directory" type="xsd:string" use="required"/>
<xsd:attribute name="local-working-directory" type="xsd:string"/>
<xsd:attribute name="host" type="xsd:string" use="required"/>
<xsd:attribute name="password" type="xsd:string"/>
<xsd:attribute name="port" type="xsd:int" default="22"/>
<xsd:attribute name="key-file" type="xsd:string"/>
<xsd:attribute name="key-file-password" type="xsd:string"/>
<xsd:attribute name="auto-create-directories" type="xsd:boolean"/>
<xsd:attribute name="auto-delete-remote-files-on-sync" type="xsd:boolean"/>
</xsd:complexType>
</xsd:element>
</xsd:schema>

20
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SFTPFileAnnouncer.java Normal file
View File

@@ -0,0 +1,20 @@
package org.springframework.integration.sftp;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.stereotype.Component;
import java.io.File;
/**
* @author Josh Long
*
*/
@Component("sftpAnnouncer")
public class SFTPFileAnnouncer {
@ServiceActivator
public void announceFile(File file){
System.out.println( "New file from the remote host has arrived: " + file.getAbsolutePath()) ;
}
}

113
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSFTPReceipt.java Normal file
View File

@@ -0,0 +1,113 @@
package org.springframework.integration.sftp;
import org.apache.commons.lang.SystemUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.junit.Before;
import org.junit.Test;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ErrorHandler;
import java.io.File;
import java.util.logging.Logger;
/**
* This tests the API, more than the end-to-end XML to Java approach.
*
* @author Josh Long
*/
public class TestSFTPReceipt {
private static final Logger logger = Logger.getLogger(TestSFTPReceipt.class.getName());
private SFTPSessionFactory sftpSessionFactory;
private String host;
private String password;
private String user;
private String privateKeyPath;
private String privateKeyPassword;
private int port;
@Before
public void before() throws Throwable {
this.sftpSessionFactory = buildSFTPSessionFactory(this.host, this.password, this.user, this.privateKeyPath, this.privateKeyPassword, this.port);
}
@Test
public void testReceive() throws Throwable {
String localMount = SystemUtils.getUserHome() + "/local_mount";
String remoteMount = "remote_mount";
// local path
File local = new File(localMount); // obviously this is just for test. Do what you need to do in your own
// we are testing, after all
if (local.exists() && (local.list().length > 0)) {
for (File f : local.listFiles()) {
if (!f.delete()) {
logger.fine("couldn't delete " + f.getAbsolutePath());
}
}
}
Resource localDirectory = new FileSystemResource(local);
// pool
QueuedSFTPSessionPool queuedSFTPSessionPool = new QueuedSFTPSessionPool(sftpSessionFactory);
queuedSFTPSessionPool.afterPropertiesSet();
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.setPoolSize(10);
taskScheduler.setErrorHandler(new ErrorHandler() {
public void handleError(Throwable t) {
System.out.println("Error occurred: " + ExceptionUtils.getFullStackTrace(t));
}
});
taskScheduler.setWaitForTasksToCompleteOnShutdown(true);
taskScheduler.initialize();
// synchronizer
final SFTPInboundSynchronizer sftpInboundSynchronizer = new SFTPInboundSynchronizer();
sftpInboundSynchronizer.setLocalDirectory(localDirectory);
sftpInboundSynchronizer.setRemotePath(remoteMount);
sftpInboundSynchronizer.setAutoCreatePath(true);
sftpInboundSynchronizer.setPool(queuedSFTPSessionPool);
sftpInboundSynchronizer.setShouldDeleteDownloadedRemoteFiles(false);
sftpInboundSynchronizer.setTaskScheduler(taskScheduler);
sftpInboundSynchronizer.afterPropertiesSet();
sftpInboundSynchronizer.start();
new Thread(new Runnable() {
public void run() {
try {
Thread.sleep(60 * 1000); // 1 minute
sftpInboundSynchronizer.stop();
} catch (InterruptedException e) {
// don't care
}
}
}).start();
}
private SFTPSessionFactory buildSFTPSessionFactory(String host, String pw, String usr, String pvKey, String pvKeyPass, int port)
throws Throwable {
SFTPSessionFactory sftpSessionFactory = new SFTPSessionFactory();
sftpSessionFactory.setPassword(pw);
sftpSessionFactory.setPort(port);
sftpSessionFactory.setRemoteHost(host);
sftpSessionFactory.setUser(usr);
sftpSessionFactory.setPrivateKey(pvKey);
sftpSessionFactory.setPrivateKeyPassphrase(pvKeyPass);
sftpSessionFactory.afterPropertiesSet();
return sftpSessionFactory;
}
}

58
spring-integration-sftp/src/test/resources/TestInboundSFTP.xml Normal file
View File

@@ -0,0 +1,58 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2010 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.
-->
<beans:beans
xmlns="http://www.springframework.org/schema/integration"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:lang="http://www.springframework.org/schema/lang"
xmlns:sftp="http://www.springframework.org/schema/integration/sftp"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd
http://www.springframework.org/schema/tool http://www.springframework.org/schema/tool/spring-tool-3.0.xsd
http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-3.0.xsd
http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp-2.0.xsd">
<context:component-scan base-package="org.springframework.integration.sftp"/>
<channel id="inboundFilesChannel"/>
<sftp:inbound-channel-adapter username="user" remote-directory="/home/user/Desktop/in" password="password"
host="host" channel="inboundFilesChannel">
<poller>
<interval-trigger interval="10"/>
</poller>
</sftp:inbound-channel-adapter>
<sftp:outbound-channel-adapter
key-file="/home/user/user.pem"
remote-directory="remote_mount_key"
channel="inboundFilesChannel"
username="ubuntu"
host="siteonec2usingubuntuami.com"/>
</beans:beans>

55
spring-integration-sftp/src/test/resources/TestOutboundSFTP.xml Normal file
View File

@@ -0,0 +1,55 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2010 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.
-->
<beans:beans
xmlns="http://www.springframework.org/schema/integration"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:lang="http://www.springframework.org/schema/lang"
xmlns:sftp="http://www.springframework.org/schema/integration/sftp"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd
http://www.springframework.org/schema/tool http://www.springframework.org/schema/tool/spring-tool-3.0.xsd
http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-3.0.xsd
http://www.springframework.org/schema/integration/sftp
http://www.springframework.org/schema/integration/sftp/spring-integration-sftp-2.0.xsd">
<context:component-scan base-package="org.springframework.integration.sftp"/>
<channel id="inboundFilesChannel"/>
<sftp:outbound-channel-adapter
key-file="/home/user/user.pem"
remote-directory="/home/ubuntu/remote_mount_key"
auto-create-directories="true"
channel="inboundFilesChannel"
username="ubuntu"
host="siteonec2usingubuntuami.com"
/>
<service-activator input-channel="inboundFilesChannel" ref="sftpAnnouncer"/>
</beans:beans>

15
spring-integration-sftp/template.mf Normal file
View File

@@ -0,0 +1,15 @@
Bundle-SymbolicName: org.springframework.integration.sftp
Bundle-Name: Spring Integration SFTP Support
Bundle-Vendor: SpringSource
Bundle-ManifestVersion: 2
Import-Template:
org.apache.commons.logging;version="[1.1.1, 2.0.0)",
org.apache.commons.lang.*;version="[2.5.0, 3.0.0)",
org.springframework.integration.*;version="[2.0.0, 2.0.1)",
org.springframework.beans.*;version="[3.0.0, 4.0.0)",
org.springframework.context;version="[3.0.0, 4.0.0)",
org.springframework.core.*;version="[3.0.0, 4.0.0)",
org.springframework.util;version="[3.0.0, 4.0.0)",
org.jivesoftware.*;version="[3.1.0, 4.0.0)",
javax.*;version="0",
org.w3c.dom.*;version="0"