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

INT-1614 all logic for inbound adapters is now in a base class

Browse Source
This commit is contained in:
Mark Fisher
2010-11-22 00:31:50 -05:00
parent 4476620094
commit c9a1af9970
8 changed files with 158 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteInboundChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,87 @@
/*
* 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.file.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.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
* @author Mark Fisher
* @since 2.0
*/
public abstract class AbstractRemoteInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
// build the SessionFactory
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.file.remote.session.CachingSessionFactory");
sessionFactoryBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
// build the InboundFileSynchronizer
BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
this.getInboundFileSynchronizerClassname());
synchronizerBuilder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "remote-directory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "delete-remote-files");
// configure the FileListFilter if relevant attributes are present
String fileNamePattern = element.getAttribute("filename-pattern");
String filter = element.getAttribute("filter");
boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern);
boolean hasFilter = StringUtils.hasText(filter);
if (hasFileNamePattern || hasFilter) {
if (!(hasFileNamePattern ^ hasFilter)) {
throw new BeanDefinitionStoreException("at most one of 'filename-pattern' or 'filter' " +
"is allowed on remote file inbound adapter");
}
}
if (hasFileNamePattern) {
BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
this.getSimplePatternFileListFilterClassname());
filterBuilder.addConstructorArgValue(fileNamePattern);
synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
}
else if (hasFilter) {
synchronizerBuilder.addPropertyReference("filter", filter);
}
// build the MessageSource
BeanDefinitionBuilder messageSourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(this.getMessageSourceClassname());
messageSourceBuilder.addConstructorArgValue(synchronizerBuilder.getBeanDefinition());
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "local-directory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "auto-create-directories");
return messageSourceBuilder.getBeanDefinition();
}
protected abstract String getMessageSourceClassname();
protected abstract String getInboundFileSynchronizerClassname();
protected abstract String getSimplePatternFileListFilterClassname();
}

35
spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java
View File

@@ -55,35 +55,37 @@ import org.springframework.util.Assert;
public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends MessageProducerSupport implements MessageSource<File> {
/**
* Should the endpoint attempt to create the local directory and/or the remote directory?
* Should the endpoint attempt to create the local directory?
*/
protected volatile boolean autoCreateDirectories = true;
private volatile boolean autoCreateDirectories = true;
/**
* An implementation that will handle the chores of actually connecting to and synching up
* An implementation that will handle the chores of actually connecting to and synchronizing
* the remote file system with the local one, in an inbound direction.
*/
protected volatile AbstractInboundFileSynchronizer<F> synchronizer;
private final AbstractInboundFileSynchronizer<F> synchronizer;
/**
* Directory to which things should be synched locally.
* Directory to which things should be synchronized locally.
*/
protected volatile File localDirectory;
private volatile File localDirectory;
/**
* The actual {@link FileReadingMessageSource} that monitors the local filesystem once files are synched.
* The actual {@link FileReadingMessageSource} that monitors the local file system once files are synchronized.
*/
protected volatile FileReadingMessageSource fileSource;
private final FileReadingMessageSource fileSource = new FileReadingMessageSource();;
public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer<F> synchronizer) {
Assert.notNull(synchronizer, "synchronizer must not be null");
this.synchronizer = synchronizer;
}
public void setAutoCreateDirectories(boolean autoCreateDirectories) {
this.autoCreateDirectories = autoCreateDirectories;
}
public void setSynchronizer(AbstractInboundFileSynchronizer<F> synchronizer) {
this.synchronizer = synchronizer;
}
public void setLocalDirectory(File localDirectory) {
this.localDirectory = localDirectory;
}
@@ -102,11 +104,8 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends M
throw new FileNotFoundException(this.localDirectory.getName());
}
}
// Forwards files once they ultimately appear in the {@link #localDirectory}.
this.fileSource = new FileReadingMessageSource();
this.fileSource.setFilter(this.buildFilter());
this.fileSource.setDirectory(this.localDirectory);
this.fileSource.setFilter(this.buildFilter());
this.fileSource.afterPropertiesSet();
this.synchronizer.afterPropertiesSet();
}
@@ -114,8 +113,8 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends M
throw e;
}
catch (Exception e) {
throw new MessagingException("Failure during initialization of MessageSource for: "
+ this.getComponentType(), e);
throw new MessagingException(
"Failure during initialization of MessageSource for: " + this.getComponentType(), e);
}
}

60
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java
View File

@@ -16,60 +16,30 @@
package org.springframework.integration.ftp.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.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
import org.springframework.integration.file.config.AbstractRemoteInboundChannelAdapterParser;
/**
* @author Oleg Zhurakousky
* @author Mark Fisher
* @since 2.0
*/
public class FtpInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
public class FtpInboundChannelAdapterParser extends AbstractRemoteInboundChannelAdapterParser {
private static final String BASE_PACKAGE = "org.springframework.integration.ftp";
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder messageSourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource");
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "auto-create-directories");
protected String getMessageSourceClassname() {
return BASE_PACKAGE + ".inbound.FtpInboundFileSynchronizingMessageSource";
}
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.file.remote.session.CachingSessionFactory");
sessionFactoryBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
@Override
protected String getInboundFileSynchronizerClassname() {
return BASE_PACKAGE + ".inbound.FtpInboundFileSynchronizer";
}
BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer");
synchronizerBuilder.addConstructorArgValue(sessionFactoryBuilder.getBeanDefinition());
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "remote-directory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "delete-remote-files");
String fileNamePattern = element.getAttribute("filename-pattern");
String filter = element.getAttribute("filter");
boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern);
boolean hasFilter = StringUtils.hasText(filter);
if (hasFileNamePattern || hasFilter) {
if (!(hasFileNamePattern ^ hasFilter)) {
throw new BeanDefinitionStoreException("at most one of 'filename-pattern' or 'filter' " +
"is allowed on FTP inbound adapter");
}
}
if (hasFileNamePattern) {
BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter");
filterBuilder.addConstructorArgValue(fileNamePattern);
synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
}
else if (hasFilter) {
synchronizerBuilder.addPropertyReference("filter", filter);
}
messageSourceBuilder.addPropertyValue("synchronizer", synchronizerBuilder.getBeanDefinition());
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "local-directory");
return messageSourceBuilder.getBeanDefinition();
@Override
protected String getSimplePatternFileListFilterClassname() {
return BASE_PACKAGE + ".filters.FtpSimplePatternFileListFilter";
}
}

8
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizingMessageSource.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.ftp.inbound;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource;
/**
@@ -25,10 +26,17 @@ import org.springframework.integration.file.remote.synchronizer.AbstractInboundF
*
* @author Iwein Fuld
* @author Josh Long
* @since 2.0
*/
public class FtpInboundFileSynchronizingMessageSource extends AbstractInboundFileSynchronizingMessageSource<FTPFile> {
public FtpInboundFileSynchronizingMessageSource( AbstractInboundFileSynchronizer<FTPFile> synchronizer) {
super(synchronizer);
}
public String getComponentType() {
return "ftp:inbound-channel-adapter";
}
}

10
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTest.java
View File

@@ -73,15 +73,13 @@ public class FtpInboundRemoteFileSystemSynchronizerTest {
ftpSessionFactory.setHost("foo.com");
ftpSessionFactory.setRemoteWorkingDirectory("remote-test-dir");
FtpInboundFileSynchronizingMessageSource ms =
new FtpInboundFileSynchronizingMessageSource();
FtpInboundFileSynchronizer synchronizer = spy(new FtpInboundFileSynchronizer(ftpSessionFactory));
synchronizer.setDeleteRemoteFiles(true);
synchronizer.setRemoteDirectory("remote-test-dir");
synchronizer.setFilter(new FtpRegexPatternFileListFilter(".*\\.test$"));
ms.setSynchronizer(synchronizer);
FtpInboundFileSynchronizingMessageSource ms =
new FtpInboundFileSynchronizingMessageSource(synchronizer);
ms.setAutoCreateDirectories(true);
ms.setLocalDirectory(localDirectoy);
@@ -95,7 +93,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTest {
Message<File> nothing = ms.receive();
assertNull(nothing);
// two times becouse on teh third receive (above) the internal queue will be empty, so it will attempt
// two times because on the third receive (above) the internal queue will be empty, so it will attempt
verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectoy);
assertTrue(new File("test/a.test").exists());

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

@@ -16,63 +16,32 @@
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;
import org.springframework.integration.file.config.AbstractRemoteInboundChannelAdapterParser;
/**
* Parser for 'sftp:inbound-channel-adapter'
*
* @author Oleg Zhurakousky
* @author Mark Fisher
* @since 2.0
*/
public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
public class SftpInboundChannelAdapterParser extends AbstractRemoteInboundChannelAdapterParser {
private static final String BASE_PACKAGE = "org.springframework.integration.sftp";
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
String sessionFactoryName = element.getAttribute("session-factory");
String fileNamePattern = element.getAttribute("filename-pattern");
String filter = element.getAttribute("filter");
boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern);
boolean hasFilter = StringUtils.hasText(filter);
if (hasFileNamePattern || hasFilter) {
if (!(hasFileNamePattern ^ hasFilter)) {
throw new BeanDefinitionStoreException("at most one of 'filename-pattern' or 'filter' " +
"is allowed on SFTP inbound adapter");
}
}
BeanDefinitionBuilder sessionFactoryBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.file.remote.session.CachingSessionFactory");
sessionFactoryBuilder.addConstructorArgReference(sessionFactoryName);
String sessionPollName = BeanDefinitionReaderUtils.registerWithGeneratedName(
sessionFactoryBuilder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer");
synchronizerBuilder.addConstructorArgReference(sessionPollName);
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "remote-directory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "delete-remote-files");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(synchronizerBuilder, element, "filter");
BeanDefinitionBuilder messageSourceBuilder = BeanDefinitionBuilder.rootBeanDefinition(
"org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource");
messageSourceBuilder.addPropertyValue("synchronizer", synchronizerBuilder.getBeanDefinition());
if (hasFileNamePattern) {
if (parserContext.getRegistry().containsBeanDefinition(fileNamePattern)) {
messageSourceBuilder.addPropertyReference("filenamePattern", fileNamePattern);
}
else {
messageSourceBuilder.addPropertyValue("filenamePattern", fileNamePattern);
}
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "auto-create-directories");
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "local-directory");
return messageSourceBuilder.getBeanDefinition();
protected String getMessageSourceClassname() {
return BASE_PACKAGE + ".inbound.SftpInboundFileSynchronizingMessageSource";
}
@Override
protected String getInboundFileSynchronizerClassname() {
return BASE_PACKAGE + ".inbound.SftpInboundFileSynchronizer";
}
@Override
protected String getSimplePatternFileListFilterClassname() {
return BASE_PACKAGE + ".filters.SftpSimplePatternFileListFilter";
}
}

47
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizingMessageSource.java
View File

@@ -16,15 +16,11 @@
package org.springframework.integration.sftp.inbound;
import java.io.FileNotFoundException;
import java.util.regex.Pattern;
import org.springframework.integration.MessagingException;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource;
import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.ChannelSftp.LsEntry;
/**
* a {@link org.springframework.integration.core.MessageSource} implementation for SFTP
@@ -35,46 +31,13 @@ import com.jcraft.jsch.ChannelSftp;
*/
public class SftpInboundFileSynchronizingMessageSource extends AbstractInboundFileSynchronizingMessageSource<ChannelSftp.LsEntry> {
private volatile Pattern filenamePattern;
public void setFilenamePattern(Pattern filenamePattern) {
this.filenamePattern = filenamePattern;
public SftpInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer<LsEntry> synchronizer) {
super(synchronizer);
}
public String getComponentType() {
return "sftp:inbound-channel-adapter";
}
@Override
protected void 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.mkdirs();
}
else {
throw new FileNotFoundException(this.localDirectory.getName());
}
}
// Forwards files once they appear in the {@link #localDirectory}.
this.fileSource = new FileReadingMessageSource();
this.fileSource.setDirectory(this.localDirectory);
this.fileSource.afterPropertiesSet();
if (this.filenamePattern != null) {
SftpRegexPatternFileListFilter filter = new SftpRegexPatternFileListFilter(this.filenamePattern);
this.synchronizer.setFilter(filter);
}
}
catch (RuntimeException e) {
throw e;
}
catch (Exception e) {
throw new MessagingException("Failure during initialization of MessageSource for: " + this.getComponentType(), e);
}
}
}

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

@@ -75,17 +75,14 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
ftpSessionFactory.setPassword("frog");
ftpSessionFactory.setHost("foo.com");
SftpInboundFileSynchronizingMessageSource ms =
new SftpInboundFileSynchronizingMessageSource();
SftpInboundFileSynchronizer synchronizer = spy(new SftpInboundFileSynchronizer(ftpSessionFactory));
synchronizer.setDeleteRemoteFiles(true);
synchronizer.setRemoteDirectory("remote-test-dir");
synchronizer.setFilter(new SftpRegexPatternFileListFilter(".*\\.test$"));
ms.setSynchronizer(synchronizer);
ms.setAutoCreateDirectories(true);
SftpInboundFileSynchronizingMessageSource ms =
new SftpInboundFileSynchronizingMessageSource(synchronizer);
ms.setAutoCreateDirectories(true);
ms.setLocalDirectory(localDirectoy);
ms.afterPropertiesSet();
Message<File> atestFile = ms.receive();
@@ -97,7 +94,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
Message<File> nothing = ms.receive();
assertNull(nothing);
// two times becouse on teh third receive (above) the internal queue will be empty, so it will attempt
// two times because on the third receive (above) the internal queue will be empty, so it will attempt
verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectoy);
assertTrue(new File("test/a.test").exists());