diff --git a/pom.xml b/pom.xml index 44d941cb78..fa8870ea3b 100644 --- a/pom.xml +++ b/pom.xml @@ -28,6 +28,8 @@ spring-integration-ws spring-integration-xml spring-integration-xmpp + spring-integration-ftp + spring-integration-sftp UTF-8 diff --git a/spring-integration-ftp/pom.xml b/spring-integration-ftp/pom.xml new file mode 100644 index 0000000000..5694b842e0 --- /dev/null +++ b/spring-integration-ftp/pom.xml @@ -0,0 +1,99 @@ + + + 4.0.0 + + org.springframework.integration + spring-integration-parent + 2.0.0.BUILD-SNAPSHOT + + org.springframework.integration + spring-integration-ftp + jar + Spring Integration FTP Support + + + javax.activation + activation + 1.1.1 + true + + + org.springframework.integration + spring-integration-file + ${project.version} + + + commons-net + commons-net + 2.0 + + + cglib + cglib-nodep + ${cglib.version} + test + + + org.easymock + easymock + ${org.easymock.version} + test + + + org.easymock + easymockclassextension + ${org.easymock.version} + test + + + junit + junit + ${junit.version} + test + + + org.springframework + spring-context-support + ${org.springframework.version} + compile + + + org.springframework + spring-test + ${org.springframework.version} + test + + + org.springframework.integration + spring-integration-stream + ${project.version} + compile + + + org.springframework.integration + spring-integration-core + ${project.version} + compile + + + commons-lang + commons-lang + 2.5 + + + commons-io + commons-io + 1.4 + + + + + + diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/DefaultFTPClientFactory.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/DefaultFTPClientFactory.java new file mode 100644 index 0000000000..9c110240d6 --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/DefaultFTPClientFactory.java @@ -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 + * FTPClient.ACTIVE_LOCAL_CONNECTION_MODE (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; + } + } +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPClientFactory.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPClientFactory.java new file mode 100644 index 0000000000..c3697e8648 --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPClientFactory.java @@ -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 null. + * @throws IOException thrown when a networking IO subsystem error occurs + */ + FTPClient getClient() throws IOException; +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPClientPool.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPClientPool.java new file mode 100644 index 0000000000..8e12ee080c --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPClientPool.java @@ -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. + *

+ * The caller should NOT disconnect the client before calling this method. + *

+ * 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 null + * argument, although the endpoint implementations in + * org.springframework.integration.ftp will never pass in + * null. + */ + void releaseClient(FTPClient client); +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileSource.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileSource.java new file mode 100644 index 0000000000..2efaa37d72 --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileSource.java @@ -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, 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("^.*(? receive() { + return fileSource.receive(); + } + + public void onFailure(Message failedMessage, Throwable t) { + fileSource.onFailure(failedMessage, t); + } + + public void onSend(Message sentMessage) { + fileSource.onSend(sentMessage); + } + + public boolean isRunning() { + return synchronizer.isRunning(); + } + + public void start() { + synchronizer.start(); + } + + public void stop() { + synchronizer.stop(); + } +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPInboundSynchronizer.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPInboundSynchronizer.java new file mode 100644 index 0000000000..7e2b931057 --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPInboundSynchronizer.java @@ -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; + + +/** + * FTPInboundSynchronizer 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(); + } + } +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPMessageSourceFactoryBean.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPMessageSourceFactoryBean.java new file mode 100644 index 0000000000..be0a2adb3e --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPMessageSourceFactoryBean.java @@ -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 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 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 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; + } +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPSendingMessageHandler.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPSendingMessageHandler.java new file mode 100644 index 0000000000..3329522d51 --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPSendingMessageHandler.java @@ -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; + } +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPSendingMessageHandlerFactoryBean.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPSendingMessageHandlerFactoryBean.java new file mode 100644 index 0000000000..3b99027793 --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPSendingMessageHandlerFactoryBean.java @@ -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 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 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; + } +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java new file mode 100644 index 0000000000..123c02f748 --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java @@ -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. + *

+ * 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 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(maxPoolSize); + } + + /** + * Returns an active FTPClient connected to the configured server. When no + * clients are available in the queue a new client is created with the + * factory. + *

+ * It is possible that released clients are disconnected by the remote + * server (@see {@link FTPClient#sendNoOp()}. In this case getClient is + * called recursively to obtain a client that is still alive. For this + * reason large pools are not recommended in poor networking conditions. + */ + public FTPClient getClient() throws SocketException, IOException { + FTPClient client = pool.poll(); + + if (client == null) { + client = factory.getClient(); + } + + return prepareClient(client); + } + + /** + * Prepares the client before it is returned through + * getClient(). The default implementation will check the + * connection using a noOp and replace the client with a new one if it + * encounters a problem. + *

+ * 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); + } + } + } +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FTPNamespaceHandler.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FTPNamespaceHandler.java new file mode 100644 index 0000000000..c827e72d94 --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FTPNamespaceHandler.java @@ -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 CLIENT_MODES = new HashMap(); + + 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()); + } + } +} diff --git a/spring-integration-ftp/src/main/resources/META-INF/MANIFEST.MF b/spring-integration-ftp/src/main/resources/META-INF/MANIFEST.MF new file mode 100644 index 0000000000..e69de29bb2 diff --git a/spring-integration-ftp/src/main/resources/META-INF/spring.handlers b/spring-integration-ftp/src/main/resources/META-INF/spring.handlers new file mode 100644 index 0000000000..a69f177414 --- /dev/null +++ b/spring-integration-ftp/src/main/resources/META-INF/spring.handlers @@ -0,0 +1 @@ +http\://www.springframework.org/schema/integration/ftp=org.springframework.integration.ftp.config.FTPNamespaceHandler \ No newline at end of file diff --git a/spring-integration-ftp/src/main/resources/META-INF/spring.schemas b/spring-integration-ftp/src/main/resources/META-INF/spring.schemas new file mode 100644 index 0000000000..6442125fe9 --- /dev/null +++ b/spring-integration-ftp/src/main/resources/META-INF/spring.schemas @@ -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 diff --git a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd new file mode 100644 index 0000000000..6dd332feff --- /dev/null +++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd @@ -0,0 +1,205 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/InboundFTPFileServiceActivator.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/InboundFTPFileServiceActivator.java new file mode 100644 index 0000000000..8c1691c5d4 --- /dev/null +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/InboundFTPFileServiceActivator.java @@ -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) + 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"); + } +} diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/OutboundFTPExample.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/OutboundFTPExample.java new file mode 100644 index 0000000000..3a6b6405e7 --- /dev/null +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/OutboundFTPExample.java @@ -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"); + } +} diff --git a/spring-integration-ftp/src/test/resources/inbound-ftp-context.xml b/spring-integration-ftp/src/test/resources/inbound-ftp-context.xml new file mode 100644 index 0000000000..f0986409a8 --- /dev/null +++ b/spring-integration-ftp/src/test/resources/inbound-ftp-context.xml @@ -0,0 +1,33 @@ + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-ftp/src/test/resources/outbound-ftp-context.xml b/spring-integration-ftp/src/test/resources/outbound-ftp-context.xml new file mode 100644 index 0000000000..d80b6073b0 --- /dev/null +++ b/spring-integration-ftp/src/test/resources/outbound-ftp-context.xml @@ -0,0 +1,36 @@ + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-ftp/template.mf b/spring-integration-ftp/template.mf new file mode 100644 index 0000000000..e264796486 --- /dev/null +++ b/spring-integration-ftp/template.mf @@ -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" diff --git a/spring-integration-sftp/pom.xml b/spring-integration-sftp/pom.xml new file mode 100644 index 0000000000..9f412137d1 --- /dev/null +++ b/spring-integration-sftp/pom.xml @@ -0,0 +1,99 @@ + + + 4.0.0 + + org.springframework.integration + spring-integration-parent + 2.0.0.BUILD-SNAPSHOT + + org.springframework.integration + spring-integration-sftp + jar + Spring Integration SFTP Support + + + javax.activation + activation + 1.1.1 + true + + + org.springframework.integration + spring-integration-file + ${project.version} + + + com.jcraft + jsch + 0.1.42 + + + cglib + cglib-nodep + ${cglib.version} + test + + + org.easymock + easymock + ${org.easymock.version} + test + + + org.easymock + easymockclassextension + ${org.easymock.version} + test + + + junit + junit + ${junit.version} + test + + + org.springframework + spring-context-support + ${org.springframework.version} + compile + + + org.springframework + spring-test + ${org.springframework.version} + test + + + org.springframework.integration + spring-integration-stream + ${project.version} + compile + + + org.springframework.integration + spring-integration-core + ${project.version} + compile + + + commons-lang + commons-lang + 2.5 + + + commons-io + commons-io + 1.4 + + + + + + diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/QueuedSFTPSessionPool.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/QueuedSFTPSessionPool.java new file mode 100644 index 0000000000..aa55db4ce0 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/QueuedSFTPSessionPool.java @@ -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 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(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 session 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(); + } + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPConstants.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPConstants.java new file mode 100644 index 0000000000..9cf75d4cae --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPConstants.java @@ -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"; +} \ No newline at end of file diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPInboundSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPInboundSynchronizer.java new file mode 100644 index 0000000000..f050b2d800 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPInboundSynchronizer.java @@ -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 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); + } + } + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPMessageSource.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPMessageSource.java new file mode 100644 index 0000000000..3ad240ebf9 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPMessageSource.java @@ -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 very influenced + * by the FtpFileSource class from the Spring Integration FTP adapter. + * + * @author Josh Long + */ +public class SFTPMessageSource implements MessageSource, 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("^.*(? 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(); + } +} \ No newline at end of file diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSendingMessageHandler.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSendingMessageHandler.java new file mode 100644 index 0000000000..b6e8f14549 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSendingMessageHandler.java @@ -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); + } + } + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSession.java new file mode 100644 index 0000000000..ceb8c732d3 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSession.java @@ -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 ssh 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 foo.com work, where + * http://foo.com 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 not 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: /home/user/.ssh/known_hosts + * @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: /home/user/.ssh/known_hosts. Note + * that you may specify this or 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 + * .pem 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) { + } + } +} \ No newline at end of file diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSessionFactory.java new file mode 100644 index 0000000000..d47e9e1d97 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSessionFactory.java @@ -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, 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 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; + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSessionPool.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSessionPool.java new file mode 100644 index 0000000000..c8cff47a4e --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPSessionPool.java @@ -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 (session + * ,channel).disconnect() ? + * + * @param session the session to relinquish / renew + */ + void release(SFTPSession session); +} \ No newline at end of file diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSendingConsumerFactoryBean.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSendingConsumerFactoryBean.java new file mode 100644 index 0000000000..3e2300e073 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSendingConsumerFactoryBean.java @@ -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 { + 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 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; + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSourceFactoryBean.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSourceFactoryBean.java new file mode 100644 index 0000000000..274db9bfea --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSourceFactoryBean.java @@ -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 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 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 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; + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPNamespaceHandler.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPNamespaceHandler.java new file mode 100644 index 0000000000..a319076c58 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPNamespaceHandler.java @@ -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()); + } + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPSessionUtils.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPSessionUtils.java new file mode 100644 index 0000000000..990222e852 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPSessionUtils.java @@ -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 anything + */ + 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; + } +} diff --git a/spring-integration-sftp/src/main/resources/JSCH_LICENSE.txt b/spring-integration-sftp/src/main/resources/JSCH_LICENSE.txt new file mode 100644 index 0000000000..644a17459b --- /dev/null +++ b/spring-integration-sftp/src/main/resources/JSCH_LICENSE.txt @@ -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. \ No newline at end of file diff --git a/spring-integration-sftp/src/main/resources/META-INF/MANIFEST.MF b/spring-integration-sftp/src/main/resources/META-INF/MANIFEST.MF new file mode 100644 index 0000000000..e69de29bb2 diff --git a/spring-integration-sftp/src/main/resources/META-INF/spring.handlers b/spring-integration-sftp/src/main/resources/META-INF/spring.handlers new file mode 100644 index 0000000000..6889a5f16f --- /dev/null +++ b/spring-integration-sftp/src/main/resources/META-INF/spring.handlers @@ -0,0 +1 @@ +http\://www.springframework.org/schema/integration/sftp=org.springframework.integration.sftp.config.SFTPNamespaceHandler \ No newline at end of file diff --git a/spring-integration-sftp/src/main/resources/META-INF/spring.schemas b/spring-integration-sftp/src/main/resources/META-INF/spring.schemas new file mode 100644 index 0000000000..b0a4907255 --- /dev/null +++ b/spring-integration-sftp/src/main/resources/META-INF/spring.schemas @@ -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 diff --git a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.0.xsd b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.0.xsd new file mode 100644 index 0000000000..0316c0122d --- /dev/null +++ b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.0.xsd @@ -0,0 +1,117 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + poller 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: + +

    +
  1. Password authentication: using this opton, authenticatio is done using a username and a password.
    2. +
  2. 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.
    3. +
+ ]]> + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SFTPFileAnnouncer.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SFTPFileAnnouncer.java new file mode 100644 index 0000000000..be476a785b --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SFTPFileAnnouncer.java @@ -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()) ; + } + +} diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSFTPReceipt.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSFTPReceipt.java new file mode 100644 index 0000000000..b599782058 --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSFTPReceipt.java @@ -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; + } +} diff --git a/spring-integration-sftp/src/test/resources/TestInboundSFTP.xml b/spring-integration-sftp/src/test/resources/TestInboundSFTP.xml new file mode 100644 index 0000000000..eee6bc37f5 --- /dev/null +++ b/spring-integration-sftp/src/test/resources/TestInboundSFTP.xml @@ -0,0 +1,58 @@ + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-sftp/src/test/resources/TestOutboundSFTP.xml b/spring-integration-sftp/src/test/resources/TestOutboundSFTP.xml new file mode 100644 index 0000000000..78a5ecc7f8 --- /dev/null +++ b/spring-integration-sftp/src/test/resources/TestOutboundSFTP.xml @@ -0,0 +1,55 @@ + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-sftp/template.mf b/spring-integration-sftp/template.mf new file mode 100644 index 0000000000..875458a914 --- /dev/null +++ b/spring-integration-sftp/template.mf @@ -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"