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

INT-1631 first round removed factory beans in favor of bootstrapping required components in the parser. Eliminated a lot of repetative code, fixed the schema with regard to what is required and optional attributes, added inbound and outbound ignored samples

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-11-17 19:46:32 -05:00
parent 46b0b56391
commit a196501ca9
29 changed files with 687 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-sftp/.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
.test.txt

0
spring-integration-sftp/local-test-dir/bar.txt Normal file
View File

0
spring-integration-sftp/local-test-dir/foo.txt Normal file
View File

1
spring-integration-sftp/local-test-dir/readme.txt Normal file
View File

@@ -0,0 +1 @@
Don't delete. This directory is used by test cases

0
spring-integration-sftp/local-test-dir/xyz.txt Normal file
View File

1
spring-integration-sftp/remote-test-dir/readme.txt Normal file
View File

@@ -0,0 +1 @@
Don't delete. This directory is used by test cases

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

@@ -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();
}
}

192
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundSynchronizingMessageSourceFactoryBean.java
View File

@@ -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<SftpInboundSynchronizingMessageSource> 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<ChannelSftp.LsEntry> 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<ChannelSftp.LsEntry> 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<ChannelSftp.LsEntry> compositeFtpFileListFilter = new CompositeFileListFilter<ChannelSftp.LsEntry>();
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();
}
}

67
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpNamespaceHandler.java
View File

@@ -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();
}
}
}

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

@@ -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();
}
}

86
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpSendingMessageHandlerFactoryBean.java
View File

@@ -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<SftpSendingMessageHandler> {
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<? extends SftpSendingMessageHandler> getObjectType() {
return SftpSendingMessageHandler.class;
}
public boolean isSingleton() {
return false;
}
}

44
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java
View File

@@ -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<ChannelSftp.LsEntry> {
@@ -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<ChannelSftp.LsEntry> 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);
}
}

140
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizingMessageSource.java
View File

@@ -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<ChannelSftp.LsEntry, SftpInboundSynchronizer> {
public class SftpInboundSynchronizingMessageSource extends
AbstractInboundRemoteFileSystemSynchronizingMessageSource<ChannelSftp.LsEntry, SftpInboundSynchronizer> {
/**
* 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<File> 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<File> 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);
}
}
}

142
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandler.java
View File

@@ -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<String> 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<String>(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);
}
}
}

143
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java
View File

@@ -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<SftpSession> 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<SftpSession>(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 <code>session</code> 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<SftpSession>(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);
}
}
}

2
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java
View File

@@ -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;
}

1
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionFactory.java
View File

@@ -26,6 +26,7 @@ import org.springframework.util.StringUtils;
*
* @author Josh Long
* @author Mario Gray
* @since 2.0
*/
public class SftpSessionFactory {

8
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSessionPool.java
View File

@@ -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);
}

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

@@ -107,7 +107,7 @@
</xsd:attribute>
<xsd:attribute name="filename-pattern" type="xsd:string"/>
<xsd:attribute name="remote-directory" type="xsd:string" use="required"/>
<xsd:attribute name="local-directory-path" type="xsd:string"/>
<xsd:attribute name="local-directory" type="xsd:string" use="required"/>
<xsd:attribute name="auto-create-directories" type="xsd:boolean"/>
<xsd:attribute name="auto-delete-remote-files-on-sync" type="xsd:boolean"/>
</xsd:complexType>

3
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context-fail.xml
View File

@@ -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">
<poller fixed-rate="1000"/>

10
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapaterParserTests-context.xml
View File

@@ -35,10 +35,9 @@
<sftp:inbound-channel-adapter id="sftpAdapterAutoCreate"
session-factory="sftpSessionFactory"
channel="requestChannel"
filter="filter"
filename-pattern="foo.txt"
remote-directory="ftp://foo"
local-directory-path="file:src/main/resources"
remote-directory="/foo"
local-directory="file:local-test-dir"
auto-create-directories="false"
auto-delete-remote-files-on-sync="false">
<poller fixed-rate="1000"/>
@@ -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">
<poller fixed-rate="1000"/>

3
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/MessageHistory-context.xml
View File

@@ -33,7 +33,8 @@
<sftp:inbound-channel-adapter id="sftpAdapter"
session-factory="sftpSessionFactory"
auto-startup="false"
remote-directory="sftp://hello"
remote-directory="/hello"
local-directory="file:local-test-dir"
channel="inboundFilesChannel"
filename-pattern=".*?jpg"
auto-create-directories="true"

6
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapaterParserTests.java
View File

@@ -50,11 +50,11 @@ public class OutboundChannelAdapaterParserTests {
Expression remoteDirectoryExpression = (Expression) TestUtils.getPropertyValue(handler, "remoteDirectoryExpression");
assertNotNull(remoteDirectoryExpression);
assertTrue(remoteDirectoryExpression instanceof LiteralExpression);
assertEquals(context.getBean("fileNameGenerator"), TestUtils.getPropertyValue(handler, "fileNameGenerator"));
assertEquals(context.getBean("fileNameGenerator"), TestUtils.getPropertyValue(handler, "filenameGenerator"));
assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset"));
assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolder"));
assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolderFile"));
QueuedSftpSessionPool clientPoll = (QueuedSftpSessionPool) TestUtils.getPropertyValue(handler, "pool");
QueuedSftpSessionPool clientPoll = (QueuedSftpSessionPool) TestUtils.getPropertyValue(handler, "sessionPool");
SftpSessionFactory clientFactory = (SftpSessionFactory) TestUtils.getPropertyValue(clientPoll, "sftpSessionFactory");
assertEquals("localhost", TestUtils.getPropertyValue(clientFactory, "host"));
assertEquals(2222, TestUtils.getPropertyValue(clientFactory, "port"));
@@ -72,7 +72,7 @@ public class OutboundChannelAdapaterParserTests {
SpelExpression remoteDirectoryExpression = (SpelExpression) TestUtils.getPropertyValue(handler, "remoteDirectoryExpression");
assertNotNull(remoteDirectoryExpression);
assertEquals("'foo' + '/' + 'bar'", remoteDirectoryExpression.getExpressionString());
assertEquals(context.getBean("fileNameGenerator"), TestUtils.getPropertyValue(handler, "fileNameGenerator"));
assertEquals(context.getBean("fileNameGenerator"), TestUtils.getPropertyValue(handler, "filenameGenerator"));
assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset"));
assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolder"));
assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolderFile"));

44
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample-ignored-context.xml Normal file
View File

@@ -0,0 +1,44 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp-2.0.xsd">
<bean id="sftpSessionFactory" class="org.springframework.integration.sftp.session.SftpSessionFactory">
<property name="host" value="localhost"/>
<!-- <property name="knownHosts" value="local, foo.com, bar.foo"/>-->
<property name="privateKey" value="file:/Users/ozhurakousky/.ssh/sftp_rsa"/>
<property name="privateKeyPassphrase" value="seva@1994"/>
<!-- <property name="password" value="hello"/>-->
<property name="port" value="22"/>
<property name="user" value="ozhurakousky"/>
</bean>
<int-sftp:inbound-channel-adapter id="sftpInbondAdapter"
channel="receiveChannel"
session-factory="sftpSessionFactory"
local-directory="file:/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-sftp/local-test-dir"
remote-directory="/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-sftp/remote-test-dir"
auto-startup="true"
auto-create-directories="false"
auto-delete-remote-files-on-sync="false"
filename-pattern=".*\.txt$">
<int:poller fixed-rate="10000" max-messages-per-poll="10"/>
</int-sftp:inbound-channel-adapter>
<int:channel id="receiveChannel">
<int:queue/>
</int:channel>
<bean id="filter" class="org.springframework.integration.sftp.filters.SftpPatternMatchingFileListFilter">
<constructor-arg value=".*\.txt$"/>
</bean>
</beans>

50
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpInboundReceiveSample.java Normal file
View File

@@ -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<File> message = MessageBuilder.withPayload(file).build();
// MessageChannel inputChannel = ac.getBean("inputChannel", MessageChannel.class);
// inputChannel.send(message);
// Thread.sleep(2000);
// }
// System.out.println("Done");
// ac.stop();
}
}

40
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundTransferSample-ignored.xml Normal file
View File

@@ -0,0 +1,40 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp-2.0.xsd">
<bean id="sftpSessionFactory" class="org.springframework.integration.sftp.session.SftpSessionFactory">
<property name="host" value="localhost"/>
<!-- <property name="knownHosts" value="local, foo.com, bar.foo"/>-->
<property name="privateKey" value="file:/Users/ozhurakousky/.ssh/sftp_rsa"/>
<property name="privateKeyPassphrase" value="seva@1994"/>
<!-- <property name="password" value="hello"/>-->
<property name="port" value="22"/>
<property name="user" value="ozhurakousky"/>
</bean>
<int:channel id="inputChannel"/>
<int-sftp:outbound-channel-adapter id="sftpOutboundAdapter"
session-factory="sftpSessionFactory"
channel="inputChannel"
charset="UTF-8"
remote-directory="/Users/ozhurakousky/workspace-sts-2.3.3.M2/si/spring-integration/spring-integration-sftp/remote-test-dir"/>
<!-- <int-sftp:outbound-channel-adapter id="sftpOutboundAdapterWithExpression"-->
<!-- session-factory="sftpSessionFactory"-->
<!-- channel="inputChannel"-->
<!-- charset="UTF-8"-->
<!-- filename-generator="fileNameGenerator"-->
<!-- remote-directory-expression="'foo' + '/' + 'bar'"/>-->
<bean id="fileNameGenerator" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.integration.file.FileNameGenerator"/>
</bean>
</beans>

52
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundTransferSample.java Normal file
View File

@@ -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<File> message = MessageBuilder.withPayload(file).build();
MessageChannel inputChannel = ac.getBean("inputChannel", MessageChannel.class);
inputChannel.send(message);
Thread.sleep(2000);
}
System.out.println("Done");
ac.stop();
}
}

5
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java
View File

@@ -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);

10
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java
View File

@@ -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
}