diff --git a/spring-integration-sftp/.gitignore b/spring-integration-sftp/.gitignore new file mode 100644 index 0000000000..94c5648915 --- /dev/null +++ b/spring-integration-sftp/.gitignore @@ -0,0 +1 @@ +.test.txt diff --git a/spring-integration-sftp/local-test-dir/bar.txt b/spring-integration-sftp/local-test-dir/bar.txt new file mode 100644 index 0000000000..e69de29bb2 diff --git a/spring-integration-sftp/local-test-dir/foo.txt b/spring-integration-sftp/local-test-dir/foo.txt new file mode 100644 index 0000000000..e69de29bb2 diff --git a/spring-integration-sftp/local-test-dir/readme.txt b/spring-integration-sftp/local-test-dir/readme.txt new file mode 100644 index 0000000000..c94b0e60d5 --- /dev/null +++ b/spring-integration-sftp/local-test-dir/readme.txt @@ -0,0 +1 @@ +Don't delete. This directory is used by test cases \ No newline at end of file diff --git a/spring-integration-sftp/local-test-dir/xyz.txt b/spring-integration-sftp/local-test-dir/xyz.txt new file mode 100644 index 0000000000..e69de29bb2 diff --git a/spring-integration-sftp/remote-test-dir/readme.txt b/spring-integration-sftp/remote-test-dir/readme.txt new file mode 100644 index 0000000000..c94b0e60d5 --- /dev/null +++ b/spring-integration-sftp/remote-test-dir/readme.txt @@ -0,0 +1 @@ +Don't delete. This directory is used by test cases \ No newline at end of file diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java new file mode 100644 index 0000000000..a9aabacbeb --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java @@ -0,0 +1,81 @@ +/* + * Copyright 2002-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.w3c.dom.Element; + +import org.springframework.beans.BeanMetadataElement; +import org.springframework.beans.factory.BeanDefinitionStoreException; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; +/** + * Parser for 'sftp:inbound-channel-adapter' + * + * @author Oleg Zhurakousky + * @since 2.0 + * + */ +public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser { + + @Override + protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) { + String sessionFactoryName = element.getAttribute("session-factory"); + String autoStartup = element.getAttribute("auto-startup"); + + + String fileNamePattern = element.getAttribute("filename-pattern"); + String filter = element.getAttribute("filter"); + boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern); + boolean hasFilter = StringUtils.hasText(filter); + if (!(hasFileNamePattern ^ hasFilter)) { + throw new BeanDefinitionStoreException("exactly one of 'filename-pattern' or 'filter' " + + "is allowed on SFTP inbound adapter"); + } + + BeanDefinitionBuilder sessionPollBuilder = + BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.session.QueuedSftpSessionPool"); + sessionPollBuilder.addConstructorArgReference(sessionFactoryName); + sessionPollBuilder.addPropertyValue("autoStartup", autoStartup); + String sessionPollName = + BeanDefinitionReaderUtils.registerWithGeneratedName(sessionPollBuilder.getBeanDefinition(), parserContext.getRegistry()); + + BeanDefinitionBuilder synchronizerBuilder = + BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.inbound.SftpInboundSynchronizer"); + synchronizerBuilder.addConstructorArgReference(sessionPollName); + + synchronizerBuilder.addPropertyValue("autoStartup", autoStartup); + IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "remote-directory", "remotePath"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "local-directory"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "auto-delete-remote-files-on-sync", "shouldDeleteSourceFile"); + + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(synchronizerBuilder, element, "filter"); + + BeanDefinitionBuilder messageSourceBuilder = + BeanDefinitionBuilder.rootBeanDefinition("org.springframework.integration.sftp.inbound.SftpInboundSynchronizingMessageSource"); + messageSourceBuilder.addConstructorArgReference(sessionPollName); + messageSourceBuilder.addPropertyValue("synchronizer", synchronizerBuilder.getBeanDefinition()); + + IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "filename-pattern"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "auto-create-directories"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "local-directory"); + + return messageSourceBuilder.getBeanDefinition(); + } +} \ No newline at end of file diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundSynchronizingMessageSourceFactoryBean.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundSynchronizingMessageSourceFactoryBean.java deleted file mode 100644 index 16a93d6a8d..0000000000 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundSynchronizingMessageSourceFactoryBean.java +++ /dev/null @@ -1,192 +0,0 @@ -/* - * Copyright 2002-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 java.io.File; - -import org.apache.commons.lang.SystemUtils; - -import org.springframework.beans.factory.config.AbstractFactoryBean; -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.filters.CompositeFileListFilter; -import org.springframework.integration.file.filters.FileListFilter; -import org.springframework.integration.sftp.filters.SftpPatternMatchingFileListFilter; -import org.springframework.integration.sftp.inbound.SftpInboundSynchronizer; -import org.springframework.integration.sftp.inbound.SftpInboundSynchronizingMessageSource; -import org.springframework.integration.sftp.session.QueuedSftpSessionPool; -import org.springframework.integration.sftp.session.SftpSessionFactory; -import org.springframework.util.StringUtils; - -import com.jcraft.jsch.ChannelSftp; - -/** - * Factory bean to hide the fairly complex configuration possibilities for an SFTP endpoint - * - * @author Josh Long - * @author Oleg Zhurakousky - * @since 2.0 - */ -class SftpInboundSynchronizingMessageSourceFactoryBean - extends AbstractFactoryBean implements ResourceLoaderAware { - - private volatile ResourceLoader resourceLoader; - - private volatile Resource localDirectoryResource; - - private volatile String localDirectoryPath; - - private volatile String autoCreateDirectories; - - private volatile String autoDeleteRemoteFilesOnSync; - - private volatile String filenamePattern; - - private volatile FileListFilter filter; - - private volatile SftpSessionFactory sftpSessionFactory; - - private volatile boolean autoStartup; - - private String remoteDirectory; - - - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; - } - - public void setSftpSessionFactory(SftpSessionFactory sftpSessionFactory) { - this.sftpSessionFactory = sftpSessionFactory; - } - - public void setLocalDirectoryResource(Resource localDirectoryResource) { - this.localDirectoryResource = localDirectoryResource; - } - - public void setLocalDirectoryPath(String localDirectoryPath) { - this.localDirectoryPath = localDirectoryPath; - } - - public void setAutoCreateDirectories(String autoCreateDirectories) { - this.autoCreateDirectories = autoCreateDirectories; - } - - public void setAutoDeleteRemoteFilesOnSync(String autoDeleteRemoteFilesOnSync) { - this.autoDeleteRemoteFilesOnSync = autoDeleteRemoteFilesOnSync; - } - - public void setFilenamePattern(String filenamePattern) { - this.filenamePattern = filenamePattern; - } - - public void setFilter(FileListFilter filter) { - this.filter = filter; - } - - /** - * Set the remote directory to synchronize with - */ - public void setRemoteDirectory(String remoteDirectory) { - this.remoteDirectory = remoteDirectory; - } - - /** - * {@inheritDoc} - */ - public void setResourceLoader(ResourceLoader resourceLoader) { - this.resourceLoader = resourceLoader; - } - - /** - * {@inheritDoc} - */ - @Override - public Class getObjectType() { - return SftpInboundSynchronizingMessageSourceFactoryBean.class; - } - - /** - * {@inheritDoc} - * @return Fully configured SftpInboundRemoteFileSystemSynchronizingMessageSource - */ - @Override - protected SftpInboundSynchronizingMessageSource createInstance() throws Exception { - boolean autoCreatDirs = Boolean.parseBoolean(this.autoCreateDirectories); - boolean ackRemoteDir = Boolean.parseBoolean(this.autoDeleteRemoteFilesOnSync); - SftpInboundSynchronizingMessageSource sftpMsgSrc = new SftpInboundSynchronizingMessageSource(); - sftpMsgSrc.setAutoCreateDirectories(autoCreatDirs); - - // local directories - if ((this.localDirectoryResource == null) && !StringUtils.hasText(this.localDirectoryPath)) { - File tmp = SystemUtils.getJavaIoTmpDir(); - File sftpTmp = new File(tmp, "sftpInbound"); - this.localDirectoryPath = "file://" + sftpTmp.getAbsolutePath(); - } - this.localDirectoryResource = this.resourceFromString(localDirectoryPath); - - // remote predicates - CompositeFileListFilter compositeFtpFileListFilter = new CompositeFileListFilter(); - if (StringUtils.hasText(this.filenamePattern)) { - SftpPatternMatchingFileListFilter sftpFilePatternMatchingEntryListFilter = - new SftpPatternMatchingFileListFilter(filenamePattern); - compositeFtpFileListFilter.addFilter(sftpFilePatternMatchingEntryListFilter); - } - if (this.filter != null) { - compositeFtpFileListFilter.addFilter(this.filter); - } - this.filter = compositeFtpFileListFilter; - - // pools - QueuedSftpSessionPool pool = new QueuedSftpSessionPool(15, sftpSessionFactory); - pool.afterPropertiesSet(); - - SftpInboundSynchronizer sftpSync = new SftpInboundSynchronizer(); - sftpSync.setClientPool(pool); - sftpSync.setLocalDirectory(this.localDirectoryResource); - sftpSync.setShouldDeleteSourceFile(ackRemoteDir); - sftpSync.setFilter(compositeFtpFileListFilter); - sftpSync.setBeanFactory(this.getBeanFactory()); - sftpSync.setRemotePath(this.remoteDirectory); - sftpSync.afterPropertiesSet(); - sftpSync.setAutoStartup(this.autoStartup); - if (this.autoStartup){ - sftpSync.start(); - } - sftpMsgSrc.setRemotePredicate(compositeFtpFileListFilter); - sftpMsgSrc.setSynchronizer(sftpSync); - sftpMsgSrc.setClientPool(pool); - sftpMsgSrc.setRemotePath(this.remoteDirectory); - sftpMsgSrc.setLocalDirectory(this.localDirectoryResource); - sftpMsgSrc.setBeanFactory(this.getBeanFactory()); - sftpMsgSrc.afterPropertiesSet(); - sftpMsgSrc.setAutoStartup(this.autoStartup); - if (this.autoStartup){ - sftpMsgSrc.start(); - } - - return sftpMsgSrc; - } - - private Resource resourceFromString(String path) { - ResourceEditor resourceEditor = new ResourceEditor(this.resourceLoader); - resourceEditor.setAsText(path); - return (Resource) resourceEditor.getValue(); - } - -} 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 index 2bd3e9f798..23a58aa167 100644 --- 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 @@ -16,17 +16,7 @@ package org.springframework.integration.sftp.config; -import org.w3c.dom.Element; - -import org.springframework.beans.BeanMetadataElement; -import org.springframework.beans.factory.support.AbstractBeanDefinition; -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler; -import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; -import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser; -import org.springframework.integration.config.xml.IntegrationNamespaceUtils; -import org.springframework.util.StringUtils; /** * Provides namespace support for using SFTP. @@ -43,61 +33,4 @@ public class SftpNamespaceHandler extends AbstractIntegrationNamespaceHandler { registerBeanDefinitionParser("inbound-channel-adapter", new SftpInboundChannelAdapterParser()); registerBeanDefinitionParser("outbound-channel-adapter", new SftpOutboundChannelAdapterParser()); } - - - /** - * Configures an object that can receive files from a remote SFTP endpoint and broadcast their arrival to a channel. - */ - private static class SftpInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser { - - @Override - protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = - BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.config.SftpInboundSynchronizingMessageSourceFactoryBean"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "filter"); - for (String p : "auto-startup,filename-pattern,auto-create-directories,remote-directory,local-directory-path,auto-delete-remote-files-on-sync".split(",")) { - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, p); - } - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "session-factory", "sftpSessionFactory"); - return builder.getBeanDefinition(); - } - } - - - /** - * Configures an object that can take messages and send them via SFTP. - */ - private static class SftpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { - - @Override - protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = - BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.config.SftpSendingMessageHandlerFactoryBean"); - for (String p : "charset".split(",")) { - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, p); - } - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "filename-generator", "fileNameGenerator"); - String remoteDirectory = element.getAttribute("remote-directory"); - String remoteDirectoryExpression = element.getAttribute("remote-directory-expression"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "session-factory", "sftpSessionFactory"); - boolean hasLiteralRemoteDirectory = StringUtils.hasText(remoteDirectory); - boolean hasRemoteDirectoryExpression = StringUtils.hasText(remoteDirectoryExpression); - if (hasLiteralRemoteDirectory ^ hasRemoteDirectoryExpression) { - if (hasLiteralRemoteDirectory) { - builder.addPropertyValue("remoteDirectory", remoteDirectory); - } - else { - BeanDefinitionBuilder expressionDefBuilder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.config.ExpressionFactoryBean"); - expressionDefBuilder.addConstructorArgValue(remoteDirectoryExpression); - builder.addPropertyValue("remoteDirectoryExpression", expressionDefBuilder.getBeanDefinition()); - } - } - else { - parserContext.getReaderContext().error("exactly one of 'remote-directory' or 'remote-directory-expression' is required", element); - } - return builder.getBeanDefinition(); - } - } - } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpOutboundChannelAdapterParser.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpOutboundChannelAdapterParser.java new file mode 100644 index 0000000000..933fad358d --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpOutboundChannelAdapterParser.java @@ -0,0 +1,70 @@ +/* + * Copyright 2002-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.w3c.dom.Element; + +import org.springframework.beans.factory.BeanDefinitionStoreException; +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; +/** + * Parser for 'sftp:outbound-channel-adapter' + * + * @author Oleg Zhurakousky + * @since 2.0 + */ +public class SftpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { + + @Override + protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { + BeanDefinitionBuilder sessionPollBuilder = + BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.session.QueuedSftpSessionPool"); + sessionPollBuilder.addConstructorArgReference(element.getAttribute("session-factory")); + + BeanDefinitionBuilder handlerBuilder = + BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.outbound.SftpSendingMessageHandler"); + handlerBuilder.addConstructorArgValue(sessionPollBuilder.getBeanDefinition()); + IntegrationNamespaceUtils.setValueIfAttributeDefined(handlerBuilder, element, "charset"); + + String remoteDirectory = element.getAttribute("remote-directory"); + String remoteDirectoryExpression = element.getAttribute("remote-directory-expression"); + boolean hasDirectory = StringUtils.hasText(remoteDirectory); + boolean hasDirectoryExpression = StringUtils.hasText(remoteDirectoryExpression); + if (!(hasDirectory ^ hasDirectoryExpression)) { + throw new BeanDefinitionStoreException("exactly one of 'remote-directory' or 'remote-directory-expression' " + + "is required on SFTP outbound adapter"); + } + BeanDefinition expressionDef = null; + if (hasDirectory){ + expressionDef = new RootBeanDefinition("org.springframework.expression.common.LiteralExpression"); + expressionDef.getConstructorArgumentValues().addGenericArgumentValue(remoteDirectory); + } + else if (hasDirectoryExpression){ + expressionDef = new RootBeanDefinition("org.springframework.integration.config.ExpressionFactoryBean"); + expressionDef.getConstructorArgumentValues().addGenericArgumentValue(remoteDirectoryExpression); + } + handlerBuilder.addPropertyValue("remoteDirectoryExpression", expressionDef); + + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(handlerBuilder, element, "filename-generator"); + return handlerBuilder.getBeanDefinition(); + } +} \ No newline at end of file diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpSendingMessageHandlerFactoryBean.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpSendingMessageHandlerFactoryBean.java deleted file mode 100644 index 63b326399a..0000000000 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpSendingMessageHandlerFactoryBean.java +++ /dev/null @@ -1,86 +0,0 @@ -/* - * Copyright 2002-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.expression.Expression; -import org.springframework.expression.common.LiteralExpression; -import org.springframework.integration.file.FileNameGenerator; -import org.springframework.integration.sftp.outbound.SftpSendingMessageHandler; -import org.springframework.integration.sftp.session.QueuedSftpSessionPool; -import org.springframework.integration.sftp.session.SftpSessionFactory; - -/** - * Supports the construction of a MessagHandler that knows how to take inbound File objects - * and send them to a remote destination. - * - * @author Josh Long - * @author Oleg Zhurakousky - * @since 2.0 - */ -class SftpSendingMessageHandlerFactoryBean implements FactoryBean { - - private Expression remoteDirectoryExpression; - - private volatile SftpSessionFactory sftpSessionFactory; - - private volatile String charset; - - private volatile FileNameGenerator fileNameGenerator; - - public void setFileNameGenerator(FileNameGenerator fileNameGenerator) { - this.fileNameGenerator = fileNameGenerator; - } - - public void setSftpSessionFactory(SftpSessionFactory sftpSessionFactory) { - this.sftpSessionFactory = sftpSessionFactory; - } - - public void setCharset(String charset) { - this.charset = charset; - } - - public void setRemoteDirectory(String remoteDirectory) { - remoteDirectory = (remoteDirectory != null) ? remoteDirectory : ""; - this.remoteDirectoryExpression = new LiteralExpression(remoteDirectory); - } - - public void setRemoteDirectoryExpression(Expression remoteDirectoryExpression) { - this.remoteDirectoryExpression = remoteDirectoryExpression; - } - - public SftpSendingMessageHandler getObject() throws Exception { - QueuedSftpSessionPool sessionPool = new QueuedSftpSessionPool(15, sftpSessionFactory); - sessionPool.afterPropertiesSet(); - SftpSendingMessageHandler messageHandler = new SftpSendingMessageHandler(sessionPool); - messageHandler.setRemoteDirectoryExpression(this.remoteDirectoryExpression); - messageHandler.setCharset(this.charset); - messageHandler.setFileNameGenerator(this.fileNameGenerator); - messageHandler.afterPropertiesSet(); - - return messageHandler; - } - - public Class getObjectType() { - return SftpSendingMessageHandler.class; - } - - public boolean isSingleton() { - return false; - } - -} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java index fa87e59697..f6ccb8ca34 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java @@ -16,9 +16,13 @@ package org.springframework.integration.sftp.inbound; -import com.jcraft.jsch.ChannelSftp; +import java.io.File; +import java.io.FileOutputStream; +import java.io.InputStream; +import java.util.Collection; + import org.apache.commons.io.IOUtils; -import org.springframework.beans.factory.annotation.Required; + import org.springframework.core.io.Resource; import org.springframework.integration.MessagingException; import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSychronizer; @@ -26,19 +30,15 @@ import org.springframework.integration.file.synchronization.AbstractInboundRemot import org.springframework.integration.sftp.session.SftpSession; import org.springframework.integration.sftp.session.SftpSessionPool; import org.springframework.scheduling.Trigger; -import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.util.Assert; -import java.io.File; -import java.io.FileOutputStream; -import java.io.IOException; -import java.io.InputStream; -import java.util.Collection; +import com.jcraft.jsch.ChannelSftp; /** * Gandles the synchronization between a remote SFTP endpoint and a local mount. * * @author Josh Long + * @author Oleg Zhurakousky * @since 2.0 */ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSychronizer { @@ -51,26 +51,26 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych /** * the pool of {@link org.springframework.integration.sftp.session.SftpSessionPool} SFTP sessions */ - private volatile SftpSessionPool clientPool; + private final SftpSessionPool sessionPool; + + public SftpInboundSynchronizer(SftpSessionPool sessionPool){ + Assert.notNull(sessionPool, "'sessionPool' must not be null"); + this.sessionPool = sessionPool; + } public void setRemotePath(String remotePath) { this.remotePath = remotePath; } - @Required - public void setClientPool(SftpSessionPool clientPool) { - this.clientPool = clientPool; - } - @Override protected Trigger getTrigger() { - return new PeriodicTrigger(10 * 1000); + throw new UnsupportedOperationException("This method curently is not implemented"); + //return new PeriodicTrigger(10 * 1000); } @Override protected void onInit() throws Exception { - Assert.notNull(this.clientPool, "'clientPool' must not be null"); Assert.notNull(this.remotePath, "'remotePath' must not be null"); if (this.shouldDeleteSourceFile) { this.entryAcknowledgmentStrategy = new DeletionEntryAcknowledgmentStrategy(); @@ -96,7 +96,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych IOUtils.closeQuietly(in); IOUtils.closeQuietly(fileOutputStream); } - if (tmpLocalTarget.renameTo(localFile)) { + if (tmpLocalTarget.renameTo(localFile) && this.entryAcknowledgmentStrategy != null) { this.acknowledge(sftpSession, entry); } return true; @@ -112,10 +112,10 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych @Override @SuppressWarnings("unchecked") - protected void syncRemoteToLocalFileSystem() throws Exception { + protected void syncRemoteToLocalFileSystem() { SftpSession session = null; try { - session = clientPool.getSession(); + session = sessionPool.getSession(); session.start(); ChannelSftp channelSftp = session.getChannel(); Collection beforeFilter = channelSftp.ls(remotePath); @@ -128,13 +128,11 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych } } } - catch (IOException e) { + catch (Exception e) { throw new MessagingException("couldn't synchronize remote to local directory", e); } finally { - if ((session != null) && (clientPool != null)) { - clientPool.release(session); - } + this.sessionPool.release(session); } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java index c82a0891e9..32515843f5 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java @@ -15,13 +15,20 @@ */ package org.springframework.integration.sftp.inbound; -import com.jcraft.jsch.ChannelSftp; -import com.jcraft.jsch.SftpATTRS; +import java.io.File; +import java.io.FileNotFoundException; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSynchronizingMessageSource; +import org.springframework.integration.sftp.filters.SftpPatternMatchingFileListFilter; import org.springframework.integration.sftp.session.SftpSession; import org.springframework.integration.sftp.session.SftpSessionPool; -import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +import com.jcraft.jsch.ChannelSftp; +import com.jcraft.jsch.SftpATTRS; /** @@ -31,38 +38,55 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @since 2.0 */ -public class SftpInboundSynchronizingMessageSource extends AbstractInboundRemoteFileSystemSynchronizingMessageSource { +public class SftpInboundSynchronizingMessageSource extends + AbstractInboundRemoteFileSystemSynchronizingMessageSource { + /** * the pool of sessions */ - private volatile SftpSessionPool clientPool; - + private final SftpSessionPool sessionPool; /** - * the remote path on teh server + * the remote path on the server */ - private volatile String remotePath; + private volatile String remoteDirectory; - public void setClientPool(SftpSessionPool clientPool) { - this.clientPool = clientPool; + private volatile String filenamePattern; + + public SftpInboundSynchronizingMessageSource(SftpSessionPool sessionPool){ + this.sessionPool = sessionPool; + System.out.println("###### Constructing"); } - public void setRemotePath(String remotePath) { - this.remotePath = remotePath; + public void setFilenamePattern(String filenamePattern) { + this.filenamePattern = filenamePattern; } - - @Override - protected void doStart() { - this.synchronizer.start(); + + public void setRemoteDirectory(String remoteDirectory) { + this.remoteDirectory = remoteDirectory; } - - @Override - protected void doStop() { - this.synchronizer.stop(); + + public String getComponentType(){ + return "sftp:inbound-channel-adapter"; + } + + public Message receive() { + /* + * Basically keep polling from the file source untill null, + * then attempt to sync up with remote directory which should populate the file source + * if anything is there and poll on file source again and if its still null then return it. + */ + Message message = this.fileSource.receive(); + if (message == null){ + this.checkThatRemotePathExists(this.remoteDirectory); + this.synchronizer.syncRemoteToLocalFileSystem(); + message = this.fileSource.receive(); + } + return message; } /** - * there be dragons this way ... This method will check to ensure that the remote directory exists. If the directory + * This method will check to ensure that the remote directory exists. If the directory * doesnt exist, and autoCreatePath is 'true,' then this method makes a few reasonably sane attempts * to create it. Otherwise, it fails fast. * @@ -73,20 +97,25 @@ public class SftpInboundSynchronizingMessageSource extends AbstractInboundRemote private boolean checkThatRemotePathExists(String remotePath) { SftpSession session = null; ChannelSftp channelSftp = null; - try { - session = this.clientPool.getSession(); - Assert.state(session != null, "session as returned from the pool should not be null. " + "If it is, it is most likely an error in the pool implementation. "); - + session = this.sessionPool.getSession(); session.start(); channelSftp = session.getChannel(); - + } + catch (RuntimeException re) { + throw re; + } + catch (Exception e){ + throw new MessagingException("Failed to get SftpSession while checking for existance of the remote directory", e); + } + + try { SftpATTRS attrs = channelSftp.stat(remotePath); 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.autoCreateDirectories && (this.clientPool != null) && (session != null)) { + } + catch (Throwable th) { + if (this.autoCreateDirectories && (this.sessionPool != null) && (session != null)) { try { if (channelSftp != null) { channelSftp.mkdir(remotePath); @@ -95,14 +124,18 @@ public class SftpInboundSynchronizingMessageSource extends AbstractInboundRemote return true; } } - } catch (Throwable t) { - return false; + } + catch (RuntimeException re) { + throw re; } + catch (Exception e){ + throw new MessagingException("Failed to auto-create remote directory", e); + } + } - } finally { - if ((clientPool != null) && (session != null)) { - clientPool.release(session); - } + } + finally { + this.sessionPool.release(session); } return false; @@ -110,12 +143,37 @@ public class SftpInboundSynchronizingMessageSource extends AbstractInboundRemote @Override protected void onInit() { - super.onInit(); + try { + if (this.localDirectory != null && !this.localDirectory.exists()) { + if (this.autoCreateDirectories) { + if (logger.isDebugEnabled()) { + logger.debug("The '" + this.localDirectory + "' directory doesn't exist; Will create."); + } + this.localDirectory.getFile().mkdirs(); + } + else { + throw new FileNotFoundException(this.localDirectory.getFilename()); + } + } + /** + * Forwards files once they ultimately appear in the {@link #localDirectory}. + */ + this.fileSource = new FileReadingMessageSource(); + this.fileSource.setDirectory(this.localDirectory.getFile()); + this.fileSource.afterPropertiesSet(); + } + catch (RuntimeException e) { + throw e; + } + catch (Exception e) { + throw new MessagingException("Failure during initialization of MessageSource for: " + + this.getComponentType(), e); + } - this.checkThatRemotePathExists(this.remotePath); - this.synchronizer.setClientPool(this.clientPool); - } - public String getComponentType(){ - return "sftp:inbound-channel-adapter"; + if (StringUtils.hasText(this.filenamePattern)) { + SftpPatternMatchingFileListFilter sftpFilePatternMatchingEntryListFilter = + new SftpPatternMatchingFileListFilter(filenamePattern); + this.synchronizer.setFilter(sftpFilePatternMatchingEntryListFilter); + } } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandler.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandler.java index d789f88d1a..7865d22511 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandler.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandler.java @@ -23,26 +23,27 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStreamWriter; import java.nio.charset.Charset; +import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.io.IOUtils; -import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.SystemUtils; -import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.SmartLifecycle; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; import org.springframework.expression.Expression; import org.springframework.integration.Message; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessagingException; -import org.springframework.integration.core.MessageHandler; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.FileNameGenerator; +import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; import org.springframework.integration.sftp.session.SftpSession; import org.springframework.integration.sftp.session.SftpSessionPool; -import org.springframework.integration.util.AbstractExpressionEvaluator; import org.springframework.util.Assert; import org.springframework.util.FileCopyUtils; +import org.springframework.util.StringUtils; import com.jcraft.jsch.ChannelSftp; @@ -51,28 +52,37 @@ import com.jcraft.jsch.ChannelSftp; * Assumes that the payload of the inbound message is of type {@link java.io.File}. * * @author Josh Long + * @author Oleg Zhurakousky * @since 2.0 */ -public class SftpSendingMessageHandler extends AbstractExpressionEvaluator implements MessageHandler, InitializingBean { +public class SftpSendingMessageHandler extends AbstractMessageHandler implements SmartLifecycle{ private static final String TEMPORARY_FILE_SUFFIX = ".writing"; + + private final ReentrantLock lifecycleLock = new ReentrantLock(); - - private volatile SftpSessionPool pool; + private final SftpSessionPool sessionPool; + + private volatile ExpressionEvaluatingMessageProcessor directoryExpressionProcesor; private volatile Expression remoteDirectoryExpression; - private volatile FileNameGenerator fileNameGenerator = new DefaultFileNameGenerator(); + private volatile FileNameGenerator filenameGenerator = new DefaultFileNameGenerator(); private volatile File temporaryBufferFolderFile; private volatile Resource temporaryBufferFolder = new FileSystemResource(SystemUtils.getJavaIoTmpDir()); private volatile String charset = Charset.defaultCharset().name(); + + private volatile boolean started; + + private volatile boolean autoStartup = true; - public SftpSendingMessageHandler(SftpSessionPool pool) { - this.pool = pool; + public SftpSendingMessageHandler(SftpSessionPool sessionPool) { + Assert.notNull(sessionPool, "'sessionPool' must not be null"); + this.sessionPool = sessionPool; } @@ -80,8 +90,8 @@ public class SftpSendingMessageHandler extends AbstractExpressionEvaluator imple this.temporaryBufferFolder = temporaryBufferFolder; } - public void setFileNameGenerator(FileNameGenerator fileNameGenerator) { - this.fileNameGenerator = fileNameGenerator; + public void setFilenameGenerator(FileNameGenerator filenameGenerator) { + this.filenameGenerator = filenameGenerator; } public void setRemoteDirectoryExpression(Expression remoteDirectoryExpression) { @@ -92,15 +102,71 @@ public class SftpSendingMessageHandler extends AbstractExpressionEvaluator imple this.charset = charset; } - public void afterPropertiesSet() throws Exception { - Assert.notNull(this.pool, "the pool must not be null"); + protected void onInit() throws Exception { this.temporaryBufferFolderFile = this.temporaryBufferFolder.getFile(); + if (remoteDirectoryExpression != null){ + directoryExpressionProcesor = + new ExpressionEvaluatingMessageProcessor(remoteDirectoryExpression, String.class); + } + } + +// Lifecycle + + public void start() { + sessionPool.start(); + started = true; } - private File handleFileMessage(File sourceFile, File tempFile, File resultFile) throws IOException { - if (sourceFile.renameTo(resultFile)) { - return resultFile; + public void stop() { + sessionPool.stop(); + started = false; + } + + public boolean isRunning() { + return started; + } + + public int getPhase() { + return 0; + } + + public boolean isAutoStartup() { + return autoStartup; + } + + public void stop(Runnable callback) { + this.lifecycleLock.lock(); + try { + this.stop(); + callback.run(); } + finally { + this.lifecycleLock.unlock(); + } + } + + + @Override + protected void handleMessageInternal(Message message) throws Exception { + File inboundFilePayload = this.redeemForStorableFile(message); + try { + if ((inboundFilePayload != null) && inboundFilePayload.exists()) { + sendFileToRemoteEndpoint(message, inboundFilePayload); + } + } + catch (Exception e) { + throw new MessageDeliveryException(message, "Failed to transfer '" + message.getPayload() + "' to" + + " " + this.remoteDirectoryExpression.getExpressionString(), e); + } + finally { + if (inboundFilePayload != null && inboundFilePayload.exists()) { + inboundFilePayload.delete(); + } + } + } + + + private File handleFileMessage(File sourceFile, File tempFile, File resultFile) throws IOException { FileCopyUtils.copy(sourceFile, tempFile); tempFile.renameTo(resultFile); return resultFile; @@ -122,7 +188,7 @@ public class SftpSendingMessageHandler extends AbstractExpressionEvaluator imple private File redeemForStorableFile(Message message) throws MessageDeliveryException { try { Object payload = message.getPayload(); - String generateFileName = this.fileNameGenerator.generateFileName(message); + String generateFileName = this.filenameGenerator.generateFileName(message); File tempFile = new File(this.temporaryBufferFolderFile, generateFileName + TEMPORARY_FILE_SUFFIX); File resultFile = new File(this.temporaryBufferFolderFile, generateFileName); File sendableFile = null; @@ -138,46 +204,29 @@ public class SftpSendingMessageHandler extends AbstractExpressionEvaluator imple return sendableFile; } catch (Throwable th) { - throw new MessageDeliveryException(message); - } - } - - public void handleMessage(final Message message) { - Assert.notNull(this.pool, "the pool must not be null"); - File inboundFilePayload = this.redeemForStorableFile(message); - try { - if ((inboundFilePayload != null) && inboundFilePayload.exists()) { - sendFileToRemoteEndpoint(message, inboundFilePayload); - } - } - catch (Exception e) { - throw new MessageDeliveryException(message, "failed to deliver the message", e); - } - finally { - if (inboundFilePayload != null && inboundFilePayload.exists()) - inboundFilePayload.delete(); + throw new MessageDeliveryException(message, "Failed to create sendable file.", th); } } private boolean sendFileToRemoteEndpoint(Message message, File file) throws Exception { - Assert.notNull(this.pool, "pool must not be null"); - SftpSession session = this.pool.getSession(); + SftpSession session = this.sessionPool.getSession(); if (session == null) { throw new MessagingException("The session returned from the pool is null, cannot proceed."); } - session.start(); - ChannelSftp sftp = session.getChannel(); InputStream fileInputStream = null; try { + //session.start(); + ChannelSftp sftp = session.getChannel(); fileInputStream = new FileInputStream(file); String baseOfRemotePath = ""; - if (this.remoteDirectoryExpression != null) { - String result = this.evaluateExpression(this.remoteDirectoryExpression, message, String.class); - if (result != null) { + if (directoryExpressionProcesor != null){ + String result = directoryExpressionProcesor.processMessage(message); + if (StringUtils.hasText(result)){ baseOfRemotePath = result; } } - if (!StringUtils.defaultString(baseOfRemotePath).endsWith("/")) { + + if (!StringUtils.endsWithIgnoreCase(baseOfRemotePath, "/")) { baseOfRemotePath += "/"; } sftp.put(fileInputStream, baseOfRemotePath + file.getName()); @@ -185,10 +234,7 @@ public class SftpSendingMessageHandler extends AbstractExpressionEvaluator imple } finally { IOUtils.closeQuietly(fileInputStream); - if (this.pool != null) { - this.pool.release(session); - } + this.sessionPool.release(session); } } - } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java index c5fc64440c..17a9d292f4 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java @@ -18,29 +18,41 @@ package org.springframework.integration.sftp.session; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.locks.ReentrantLock; +import java.util.logging.Logger; -import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.SmartLifecycle; import org.springframework.util.Assert; +import com.jcraft.jsch.Channel; +import com.jcraft.jsch.Session; + /** * This approach - of having a SessionPool ({@link SftpSessionPool}) that has an * implementation of Queued*SessionPool ({@link QueuedSftpSessionPool}) - was * taken almost directly from the Spring Integration FTP adapter. * * @author Josh Long + * @author Oleg Zhurakousky * @since 2.0 */ -public class QueuedSftpSessionPool implements SftpSessionPool, InitializingBean { +public class QueuedSftpSessionPool implements SftpSessionPool, SmartLifecycle { + + private final ReentrantLock atomicOperationLock = new ReentrantLock(); + + private static Logger logger = Logger.getLogger(QueuedSftpSessionPool.class.getName()); public static final int DEFAULT_POOL_SIZE = 10; - private volatile Queue queue; private final SftpSessionFactory sftpSessionFactory; private final int maxPoolSize; + + private volatile boolean started; + private volatile boolean autoStartup; public QueuedSftpSessionPool(SftpSessionFactory factory) { this(DEFAULT_POOL_SIZE, factory); @@ -50,51 +62,108 @@ public class QueuedSftpSessionPool implements SftpSessionPool, InitializingBean this.sftpSessionFactory = sessionFactory; this.maxPoolSize = maxPoolSize; } - - - public void afterPropertiesSet() throws Exception { - Assert.notNull(this.sftpSessionFactory, "sftpSessionFactory must not be null"); - Assert.isTrue(this.maxPoolSize > 0, "poolSize must be greater than 0"); - this.queue = new ArrayBlockingQueue(this.maxPoolSize, true); // size, fairness to avoid starvation + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; } public SftpSession getSession() throws Exception { - SftpSession session = this.queue.poll(); - if (null == session) { - session = this.sftpSessionFactory.getSession(); - if (this.queue.size() < this.maxPoolSize) { - this.queue.add(session); + Assert.notNull(this.queue, "SftpSession is unavailable since component is not started"); + this.atomicOperationLock.lock(); + try { + SftpSession session = this.queue.poll(); + if (null == session) { + session = this.sftpSessionFactory.getSession(); + if (this.queue.size() < this.maxPoolSize) { + this.queue.add(session); + } + } + return session; + } + finally { + this.atomicOperationLock.unlock(); + } + + } + + public void release(SftpSession sftpSession) { + if (this.started){ + this.atomicOperationLock.lock(); + try { + if (queue.size() < maxPoolSize && sftpSession != null) { + queue.add(sftpSession); + } + else { + this.destroySftpSession(sftpSession); + } + } + finally { + this.atomicOperationLock.unlock(); } } - 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); + this.destroySftpSession(sftpSession); + } + } + + public void start() { + Assert.isTrue(this.maxPoolSize > 0, "poolSize must be greater than 0"); + this.atomicOperationLock.lock(); + try { + this.queue = new ArrayBlockingQueue(this.maxPoolSize, true); + } + finally { + this.atomicOperationLock.unlock(); + } + this.started = true; + } + + public void stop() { + for (SftpSession sftpSession : queue) { + this.destroySftpSession(sftpSession); } } - private void dispose(SftpSession s) { - if (s == null) { - return; + public boolean isRunning() { + return this.started; + } + public int getPhase() { + return 0; + } + + public boolean isAutoStartup() { + return this.autoStartup; + } + + public void stop(Runnable callback) { + this.atomicOperationLock.lock(); + try { + this.stop(); + callback.run(); } - 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(); + finally { + this.started = false; + this.atomicOperationLock.unlock(); } } + private void destroySftpSession(SftpSession sftpSession){ + try { + if (sftpSession != null){ + Channel channel = sftpSession.getChannel(); + if (channel.isConnected()){ + channel.disconnect(); + } + Session session = sftpSession.getSession(); + if (session.isConnected()){ + session.disconnect(); + } + } + } catch (Throwable e) { + // log and ignore + logger.warning("Exception was thrown during while destroying SftpSession. " + e); + } + } + + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java index 1efb08602c..f03276c373 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java @@ -34,6 +34,7 @@ import com.jcraft.jsch.UserInfo; * * @author Josh Long * @author Mario Gray + * @since 2.0 */ public class SftpSession { @@ -105,7 +106,6 @@ public class SftpSession { this.channel = (ChannelSftp) this.session.openChannel("sftp"); } - public ChannelSftp getChannel() { return channel; } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionFactory.java index 1987a2d727..9e5705d717 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionFactory.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionFactory.java @@ -26,6 +26,7 @@ import org.springframework.util.StringUtils; * * @author Josh Long * @author Mario Gray + * @since 2.0 */ public class SftpSessionFactory { diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionPool.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionPool.java index fc5f6abddf..1a9a8bf3a8 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionPool.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionPool.java @@ -16,13 +16,17 @@ package org.springframework.integration.sftp.session; +import org.springframework.context.Lifecycle; + /** * Holds instances of {@link SftpSession} since they are stateful * and might be in use while another operation runs. * * @author Josh Long + * @author Oleg Zhurakousky + * @since 2.0 */ -public interface SftpSessionPool { +public interface SftpSessionPool extends Lifecycle{ /** * Returns a session that can be used to connect to an sftp instance and perform operations @@ -38,5 +42,5 @@ public interface SftpSessionPool { * @param session the session to relinquish / renew */ void release(SftpSession session); - + } 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 index c1923011a4..856e372206 100644 --- 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 @@ -107,7 +107,7 @@ - + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context-fail.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context-fail.xml index 56831a6f15..c76baefb57 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context-fail.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context-fail.xml @@ -34,9 +34,8 @@ channel="requestChannel" session-factory="sftpSessionFactory" filter="filter" - filename-pattern="foo*.txt" remote-directory="ftp://foo" - local-directory-path="file:target/bar" + local-directory="file:local-test-dir" auto-create-directories="false" auto-delete-remote-files-on-sync="false"> diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context.xml index d4a8d39cbe..9975bf2c86 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context.xml @@ -35,10 +35,9 @@ @@ -48,9 +47,8 @@ channel="requestChannel" session-factory="sftpSessionFactory" filter="filter" - filename-pattern="foo.txt" - remote-directory="ftp://foo" - local-directory-path="file:target" + remote-directory="/foo" + local-directory="file:local-test-dir" auto-create-directories="true" auto-delete-remote-files-on-sync="false"> diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/MessageHistory-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/MessageHistory-context.xml index f80241ebfb..1832b634c4 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/MessageHistory-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/MessageHistory-context.xml @@ -33,7 +33,8 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample.java new file mode 100644 index 0000000000..ba2ebc6170 --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample.java @@ -0,0 +1,50 @@ +/* + * Copyright 2002-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.junit.Ignore; +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author Oleg Zhurakousy + * + */ +public class SftpInboundReceiveSample { + + @Test + @Ignore + public void testInbound() throws Exception{ + ClassPathXmlApplicationContext ac = + new ClassPathXmlApplicationContext("SftpInboundReceiveSample-ignored.xml", SftpInboundReceiveSample.class); + + System.out.println("Done"); + System.in.read(); + ac.stop(); +// File file = new File("/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-sftp/foo.txt"); +// if (file.exists()){ +// Message message = MessageBuilder.withPayload(file).build(); +// MessageChannel inputChannel = ac.getBean("inputChannel", MessageChannel.class); +// inputChannel.send(message); +// Thread.sleep(2000); +// } +// System.out.println("Done"); +// ac.stop(); + + } + +} diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundTransferSample-ignored.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundTransferSample-ignored.xml new file mode 100644 index 0000000000..ba324aa8b4 --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundTransferSample-ignored.xml @@ -0,0 +1,40 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundTransferSample.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundTransferSample.java new file mode 100644 index 0000000000..ac5cb1aa09 --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundTransferSample.java @@ -0,0 +1,52 @@ +/* + * Copyright 2002-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 java.io.File; + +import org.junit.Ignore; +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.support.MessageBuilder; + +/** + * @author Oleg Zhurakousky + * + */ +public class SftpOutboundTransferSample { + + @Test + @Ignore + public void testOutbound() throws Exception{ + ClassPathXmlApplicationContext ac = + new ClassPathXmlApplicationContext("SftpOutboundTransferSample-ignored.xml", SftpOutboundTransferSample.class); + ac.start(); + File file = new File("/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-sftp/local-test-dir/foo.txt"); + if (file.exists()){ + Message message = MessageBuilder.withPayload(file).build(); + MessageChannel inputChannel = ac.getBean("inputChannel", MessageChannel.class); + inputChannel.send(message); + Thread.sleep(2000); + } + System.out.println("Done"); + ac.stop(); + + } + +} diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java index 6fac0504df..1c5926e3bb 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java @@ -28,6 +28,7 @@ import org.springframework.core.io.Resource; import org.springframework.integration.file.synchronization.AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy; import org.springframework.integration.sftp.inbound.SftpInboundSynchronizer; import org.springframework.integration.sftp.session.SftpSession; +import org.springframework.integration.sftp.session.SftpSessionPool; import org.springframework.util.ReflectionUtils; import com.jcraft.jsch.ChannelSftp; @@ -62,7 +63,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { */ @Test public void testCopyAndRenameWhenLocalFileExists() throws Exception { - SftpInboundSynchronizer synchronizer = new SftpInboundSynchronizer(); + SftpInboundSynchronizer synchronizer = new SftpInboundSynchronizer(mock(SftpSessionPool.class)); Method method = ReflectionUtils.findMethod(synchronizer.getClass(), "copyFromRemoteToLocalDirectory", SftpSession.class, LsEntry.class, Resource.class); method.setAccessible(true); @@ -84,7 +85,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { @org.junit.Ignore @Test public void testCopyAndRenameWhenLocalFileDoesntExist() throws Exception { - SftpInboundSynchronizer synchronizer = new SftpInboundSynchronizer(); + SftpInboundSynchronizer synchronizer = new SftpInboundSynchronizer(mock(SftpSessionPool.class)); synchronizer.setEntryAcknowledgmentStrategy(mock(EntryAcknowledgmentStrategy.class)); Method method = ReflectionUtils.findMethod(synchronizer.getClass(), "copyFromRemoteToLocalDirectory", SftpSession.class, LsEntry.class, Resource.class); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java index fad303f5bd..82fbafd98c 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java @@ -49,7 +49,8 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { if (file.exists()){ file.delete(); } - SftpInboundSynchronizer syncronizer = new SftpInboundSynchronizer(); + SftpSessionPool sessionPool = mock(SftpSessionPool.class); + SftpInboundSynchronizer syncronizer = new SftpInboundSynchronizer(sessionPool); syncronizer.setLocalDirectory(new FileSystemResource(System.getProperty("java.io.tmpdir"))); syncronizer.setRemotePath("foo/bar"); @@ -57,10 +58,10 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { syncronizer.setFilter(filter); - SftpSessionPool sessionPoll = mock(SftpSessionPool.class); + SftpSession sftpSession = mock(SftpSession.class); - when(sessionPoll.getSession()).thenReturn(sftpSession); + when(sessionPool.getSession()).thenReturn(sftpSession); ChannelSftp channel = mock(ChannelSftp.class); when(channel.get((String) Mockito.any())).thenReturn(new FileInputStream(new File("template.mf"))); when(sftpSession.getChannel()).thenReturn(channel); @@ -75,13 +76,12 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { when(channel.ls("foo/bar")).thenReturn(entries); when(filter.filterFiles((Object[]) Mockito.any())).thenReturn(entries); - syncronizer.setClientPool(sessionPoll); syncronizer.setShouldDeleteSourceFile(true); syncronizer.afterPropertiesSet(); syncronizer.syncRemoteToLocalFileSystem(); - verify(sessionPoll, times(1)).getSession(); + verify(sessionPool, times(1)).getSession(); verify(sftpSession, atLeast(1)).getChannel(); // will add more validation, but for now this test is mainly to get the test coverage up }