diff --git a/.travis.yml b/.travis.yml index 188db74d62..815c5397e5 100644 --- a/.travis.yml +++ b/.travis.yml @@ -5,7 +5,14 @@ services: - mongodb - rabbitmq - redis-server +before_cache: + - rm -f $HOME/.gradle/caches/modules-2/modules-2.lock +cache: + directories: + - $HOME/.gradle/caches/ + - $HOME/.gradle/wrapper/ +install: true env: - - TERM=dumb SI_FATAL_WHEN_NO_BEANFACTORY=true -#script: -# - ./gradlew build --parallel + - TERM=dumb SI_FATAL_WHEN_NO_BEANFACTORY=true NO_REFERENCE_TASK=true +script: + - ./gradlew check --refresh-dependencies --no-daemon diff --git a/build.gradle b/build.gradle index c74c04db65..9f0fd829af 100644 --- a/build.gradle +++ b/build.gradle @@ -355,6 +355,7 @@ project('spring-integration-ftp') { compile "org.springframework:spring-context-support:$springVersion" compile("javax.activation:activation:$javaxActivationVersion", optional) testCompile "org.apache.ftpserver:ftpserver-core:$ftpServerVersion" + testCompile project(":spring-integration-file").sourceSets.test.output } } @@ -574,6 +575,7 @@ project('spring-integration-sftp') { compile "org.springframework:spring-context-support:$springVersion" compile("javax.activation:activation:$javaxActivationVersion", optional) testCompile "org.apache.sshd:sshd-core:$apacheSshdVersion" + testCompile project(":spring-integration-file").sourceSets.test.output } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java index a84063e30c..a4fdb252d6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java @@ -53,6 +53,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan registerBeanDefinitionParser("json-to-object-transformer", new JsonToObjectTransformerParser()); registerBeanDefinitionParser("payload-serializing-transformer", new PayloadSerializingTransformerParser()); registerBeanDefinitionParser("payload-deserializing-transformer", new PayloadDeserializingTransformerParser()); + registerBeanDefinitionParser("stream-transformer", new StreamTransformerParser()); registerBeanDefinitionParser("claim-check-in", new ClaimCheckInParser()); registerBeanDefinitionParser("syslog-to-map-transformer", new SyslogToMapTransformerParser()); registerBeanDefinitionParser("claim-check-out", new ClaimCheckOutParser()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StreamTransformerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StreamTransformerParser.java new file mode 100644 index 0000000000..daf4ec8a69 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StreamTransformerParser.java @@ -0,0 +1,35 @@ +/* + * Copyright 2016 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.config.xml; + +import org.springframework.integration.transformer.StreamTransformer; + +/** + * Parser for {@code } element. + * + * @author Gary Russell + * @since 4.2.12 + * + */ +public class StreamTransformerParser extends ObjectToStringTransformerParser { + + @Override + protected String getTransformerClassName() { + return StreamTransformer.class.getName(); + } + +} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd index a4d7fae646..2eba60ea34 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd @@ -1816,6 +1816,8 @@ + + @@ -2661,11 +2663,49 @@ - - + + + + + + + + + Defines a Consumer Endpoint for the + 'org.springframework.integration.transformer.StreamTransformer' + that converts an 'InputStream' payload to a byte[] or String. + Providing a 'charset' signals that the conversion to String is + required. + + + + + + + + + + + + + + + + + + Allows you to specify the Charset (e.g., US-ASCII, + ISO-8859-1, UTF-8) to be used when transforming byte[]. + None by default, meaning the payload will be byte[]. + + + + + + + @@ -2762,8 +2802,7 @@ - - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToStringTransformerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToStringTransformerParserTests.java index 76f0984499..04262ecf27 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToStringTransformerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ObjectToStringTransformerParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2016 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. @@ -21,22 +21,26 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; -import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.messaging.support.GenericMessage; -import org.springframework.integration.test.util.TestUtils; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Mark Fisher + * @author Gary Russell */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class ObjectToStringTransformerParserTests { @Autowired diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml new file mode 100644 index 0000000000..6cfd28f958 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java new file mode 100644 index 0000000000..70d81c2197 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java @@ -0,0 +1,88 @@ +/* + * Copyright 2016 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.config.xml; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.io.ByteArrayInputStream; +import java.io.InputStream; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Mark Fisher + * @author Gary Russell + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class StreamTransformerParserTests { + + @Autowired + @Qualifier("directInput") + private MessageChannel directInput; + + @Autowired + @Qualifier("charsetChannel") + private MessageChannel charsetChannel; + + @Autowired + @Qualifier("queueInput") + private MessageChannel queueInput; + + @Autowired + @Qualifier("output") + private PollableChannel output; + + @Test + public void directChannelWithStringMessage() { + this.directInput.send(new GenericMessage(new ByteArrayInputStream("foo".getBytes()))); + Message result = output.receive(0); + assertNotNull(result); + assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload()); + } + + @Test + public void queueChannelWithStringMessage() { + this.queueInput.send(new GenericMessage(new ByteArrayInputStream("foo".getBytes()))); + Message result = output.receive(3000); + assertNotNull(result); + assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload()); + } + + @Test + public void charset() { + this.charsetChannel.send(new GenericMessage(new ByteArrayInputStream("foo".getBytes()))); + Message result = output.receive(0); + assertNotNull(result); + assertEquals("foo", result.getPayload()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/SyslogTransformerParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/SyslogTransformerParserTests-context.xml index 80099946e1..828023da8d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/SyslogTransformerParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/SyslogTransformerParserTests-context.xml @@ -7,7 +7,16 @@ + + + + + + + + + diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java index 93269f7a59..f40941c0fd 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java @@ -25,6 +25,8 @@ import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.ExpressionFactoryBean; import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer; import org.springframework.util.StringUtils; /** @@ -40,7 +42,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst @Override protected final BeanMetadataElement parseSource(Element element, ParserContext parserContext) { BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition( - this.getInboundFileSynchronizerClassname()); + this.getInboundFileSynchronizerClass()); synchronizerBuilder.addConstructorArgReference(element.getAttribute("session-factory")); @@ -54,7 +56,8 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst String remoteFileSeparator = element.getAttribute("remote-file-separator"); synchronizerBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator); IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "temporary-file-suffix"); - this.configureFilter(synchronizerBuilder, element, parserContext); + FileParserUtils.configureFilter(synchronizerBuilder, element, parserContext, + getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass()); // build the MessageSource BeanDefinitionBuilder messageSourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(this.getMessageSourceClassname()); @@ -75,52 +78,12 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst return messageSourceBuilder.getBeanDefinition(); } - private void configureFilter(BeanDefinitionBuilder synchronizerBuilder, Element element, ParserContext parserContext) { - String filter = element.getAttribute("filter"); - String fileNamePattern = element.getAttribute("filename-pattern"); - String fileNameRegex = element.getAttribute("filename-regex"); - boolean hasFilter = StringUtils.hasText(filter); - boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern); - boolean hasFileNameRegex = StringUtils.hasText(fileNameRegex); - if (hasFilter || hasFileNamePattern || hasFileNameRegex) { - int count = 0; - if (hasFilter) { - count++; - } - if (hasFileNamePattern) { - count++; - } - if (hasFileNameRegex) { - count++; - } - if (count != 1) { - parserContext.getReaderContext().error("at most one of 'filename-pattern', " + - "'filename-regex', or 'filter' is allowed on remote file inbound adapter", element); - } - if (hasFilter) { - synchronizerBuilder.addPropertyReference("filter", filter); - } - else if (hasFileNamePattern) { - BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition( - this.getSimplePatternFileListFilterClassname()); - filterBuilder.addConstructorArgValue(fileNamePattern); - synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition()); - } - else if (hasFileNameRegex) { - BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition( - this.getRegexPatternFileListFilterClassname()); - filterBuilder.addConstructorArgValue(fileNameRegex); - synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition()); - } - } - } - protected abstract String getMessageSourceClassname(); - protected abstract String getInboundFileSynchronizerClassname(); + protected abstract Class getInboundFileSynchronizerClass(); - protected abstract String getSimplePatternFileListFilterClassname(); + protected abstract Class> getSimplePatternFileListFilterClass(); - protected abstract String getRegexPatternFileListFilterClassname(); + protected abstract Class> getRegexPatternFileListFilterClass(); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java new file mode 100644 index 0000000000..4dfafdc980 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java @@ -0,0 +1,76 @@ +/* + * Copyright 2016 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.config.BeanDefinition; +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.integration.core.MessageSource; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.RemoteFileOperations; +import org.springframework.util.StringUtils; + +/** + * Abstract base class for parsing remote file streaming inbound channel adapters. + * + * @author Gary Russell + * @since 4.2.12 + */ +public abstract class AbstractRemoteFileStreamingInboundChannelAdapterParser + extends AbstractPollingInboundChannelAdapterParser { + + @Override + protected final BeanMetadataElement parseSource(Element element, ParserContext parserContext) { + BeanDefinition templateDefinition = FileParserUtils.parseRemoteFileTemplate(element, parserContext, false, + getTemplateClass()); + + BeanDefinitionBuilder messageSourceBuilder = + BeanDefinitionBuilder.genericBeanDefinition(getMessageSourceClass()); + messageSourceBuilder.addConstructorArgValue(templateDefinition); + + BeanDefinition expressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression( + "remote-directory", "remote-directory-expression", parserContext, element, false); + if (expressionDef != null) { + messageSourceBuilder.addPropertyValue("remoteDirectoryExpression", expressionDef); + } + + String remoteFileSeparator = element.getAttribute("remote-file-separator"); + messageSourceBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator); + FileParserUtils.configureFilter(messageSourceBuilder, element, parserContext, + getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass()); + + String comparator = element.getAttribute("comparator"); + if (StringUtils.hasText(comparator)) { + messageSourceBuilder.addConstructorArgReference(comparator); + } + return messageSourceBuilder.getBeanDefinition(); + } + + protected abstract Class> getTemplateClass(); + + protected abstract Class> getMessageSourceClass(); + + protected abstract Class> getSimplePatternFileListFilterClass(); + + protected abstract Class> getRegexPatternFileListFilterClass(); + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java index fbfc5a27b0..df68784d15 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java @@ -22,6 +22,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.file.DefaultFileNameGenerator; +import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.remote.RemoteFileOperations; import org.springframework.util.StringUtils; @@ -88,4 +89,43 @@ public final class FileParserUtils { return templateBuilder.getBeanDefinition(); } + static void configureFilter(BeanDefinitionBuilder synchronizerBuilder, Element element, ParserContext parserContext, + Class> patternClass, Class> regexClass) { + String filter = element.getAttribute("filter"); + String fileNamePattern = element.getAttribute("filename-pattern"); + String fileNameRegex = element.getAttribute("filename-regex"); + boolean hasFilter = StringUtils.hasText(filter); + boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern); + boolean hasFileNameRegex = StringUtils.hasText(fileNameRegex); + if (hasFilter || hasFileNamePattern || hasFileNameRegex) { + int count = 0; + if (hasFilter) { + count++; + } + if (hasFileNamePattern) { + count++; + } + if (hasFileNameRegex) { + count++; + } + if (count != 1) { + parserContext.getReaderContext().error("at most one of 'filename-pattern', " + + "'filename-regex', or 'filter' is allowed on remote file inbound adapter", element); + } + if (hasFilter) { + synchronizerBuilder.addPropertyReference("filter", filter); + } + else if (hasFileNamePattern) { + BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(patternClass); + filterBuilder.addConstructorArgValue(fileNamePattern); + synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition()); + } + else if (hasFileNameRegex) { + BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(regexClass); + filterBuilder.addConstructorArgValue(fileNameRegex); + synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition()); + } + } + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java index e3f672b399..10516082e3 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java @@ -141,7 +141,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource extends Abstra .build(); } catch (IOException e) { - return new MessagingException("IOException when retrieving " + remotePath, e); + throw new MessagingException("IOException when retrieving " + remotePath, e); } } return null; diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd index 6a7069afcf..badf32f270 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.2.xsd @@ -727,15 +727,11 @@ Only files matching this regular expression will be picked up by this adapter. - + - Specify a SpEL expression which - will be used to evaluate the directory - path to where the files will be transferred - (e.g., "headers.['remote_dir'] + - '/myTransfers'"); + Allows you to specify Charset (e.g., US-ASCII, ISO-8859-1, UTF-8). [UTF-8] is default - + used when converting String payloads to bytes. diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java index 8100112e0f..cd41c9df17 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java @@ -106,7 +106,7 @@ public abstract class RemoteFileTestSupport { recursiveDelete(new File(localTemporaryFolder.getRoot(), "localTarget")); this.targetLocalDirectory = localTemporaryFolder.newFolder("localTarget"); - File file = new File(this.sourceRemoteDirectory, " " + prefix + "Source1.txt"); + File file = new File(this.sourceRemoteDirectory, prefix + "Source1.txt"); file.createNewFile(); FileOutputStream fos = new FileOutputStream(file); fos.write("source1".getBytes()); diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java index 92bec17832..b463bd4771 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java @@ -17,6 +17,8 @@ package org.springframework.integration.ftp.config; import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer; import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer; @@ -37,18 +39,18 @@ public class FtpInboundChannelAdapterParser extends AbstractRemoteFileInboundCha } @Override - protected String getInboundFileSynchronizerClassname() { - return FtpInboundFileSynchronizer.class.getName(); + protected Class getInboundFileSynchronizerClass() { + return FtpInboundFileSynchronizer.class; } @Override - protected String getSimplePatternFileListFilterClassname() { - return FtpSimplePatternFileListFilter.class.getName(); + protected Class> getSimplePatternFileListFilterClass() { + return FtpSimplePatternFileListFilter.class; } @Override - protected String getRegexPatternFileListFilterClassname() { - return FtpRegexPatternFileListFilter.class.getName(); + protected Class> getRegexPatternFileListFilterClass() { + return FtpRegexPatternFileListFilter.class; } } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java index eaf1b3c2db..c984952079 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java @@ -33,6 +33,8 @@ public class FtpNamespaceHandler extends AbstractIntegrationNamespaceHandler { @Override public void init() { registerBeanDefinitionParser("inbound-channel-adapter", new FtpInboundChannelAdapterParser()); + registerBeanDefinitionParser("inbound-streaming-channel-adapter", + new FtpStreamingInboundChannelAdapterParser()); registerBeanDefinitionParser("outbound-channel-adapter", new FtpOutboundChannelAdapterParser()); registerBeanDefinitionParser("outbound-gateway", new FtpOutboundGatewayParser()); } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java new file mode 100644 index 0000000000..f15fcd118a --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java @@ -0,0 +1,55 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.ftp.config; + +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.file.config.AbstractRemoteFileStreamingInboundChannelAdapterParser; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.RemoteFileOperations; +import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; +import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; +import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource; +import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; + +/** + * @author Gary Russell + * @since 4.2.12 + * + */ +public class FtpStreamingInboundChannelAdapterParser extends AbstractRemoteFileStreamingInboundChannelAdapterParser { + + @Override + protected Class> getTemplateClass() { + return FtpRemoteFileTemplate.class; + } + + @Override + protected Class> getMessageSourceClass() { + return FtpStreamingMessageSource.class; + } + + @Override + protected Class> getSimplePatternFileListFilterClass() { + return FtpSimplePatternFileListFilter.class; + } + + @Override + protected Class> getRegexPatternFileListFilterClass() { + return FtpRegexPatternFileListFilter.class; + } + +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java new file mode 100644 index 0000000000..d28e2ccd9e --- /dev/null +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java @@ -0,0 +1,74 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.ftp.inbound; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Comparator; +import java.util.List; + +import org.apache.commons.net.ftp.FTPFile; + +import org.springframework.integration.file.remote.AbstractFileInfo; +import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.ftp.session.FtpFileInfo; + +/** + * Message source for streaming FTP remote file contents. + * + * @author Gary Russell + * @since 4.2.12 + * + */ +public class FtpStreamingMessageSource extends AbstractRemoteFileStreamingMessageSource { + + /** + * Construct an instance with the supplied template. + * @param template the template. + */ + public FtpStreamingMessageSource(RemoteFileTemplate template) { + super(template, null); + } + + /** + * Construct an instance with the supplied template and comparator. + * Note: the comparator is applied each time the remote directory is listed + * which only occurs when the previous list is exhausted. + * @param template the template. + * @param comparator the comparator. + */ + public FtpStreamingMessageSource(RemoteFileTemplate template, + Comparator> comparator) { + super(template, comparator); + } + + @Override + public String getComponentType() { + return "ftp:inbound-streaming-channel-adapter"; + } + + @Override + protected List> asFileInfoList(Collection files) { + List> canonicalFiles = new ArrayList>(); + for (FTPFile file : files) { + canonicalFiles.add(new FtpFileInfo(file)); + } + return canonicalFiles; + } + +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java index 8f83f9f879..50bdb10272 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java @@ -146,7 +146,11 @@ public class FtpSession implements Session { public void close() { try { if (this.readingRaw.get()) { - finalizeRaw(); + if (!finalizeRaw()) { + if (this.logger.isWarnEnabled()) { + this.logger.warn("Finalize on readRaw() returned false for " + this); + } + } } this.client.disconnect(); } diff --git a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.2.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.2.xsd index 3d80508e95..912c092eb4 100644 --- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.2.xsd +++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.2.xsd @@ -22,7 +22,7 @@ - + @@ -73,38 +73,8 @@ - - - - - - - - - - - - - Identifies channel attached to this adapter. This channel where messages will be sent - to by this adapter. - - - - - - - Allows you to provide a file name pattern to - determine the file names - that need to be scanned. - This is based on - simple pattern matching (e.g., "*.txt, fo*.txt" - etc.) - - - - + + Allows you to provide a SpEL expression to @@ -120,17 +90,6 @@ - - - - Allows you to provide a Regular Expression to - determine the file names - that need to be scanned. - (e.g., - "f[o]+\.txt" etc.) - - - - - - - - - - - - Allows you to specify a reference to a - [org.springframework.integration.file.filters.FileListFilter] - bean. This filter is applied to files on the remote server and - only files that pass the filter are retrieved. - - - @@ -215,16 +158,29 @@ - + + + + + + + + + + Configures a 'SourcePollingChannelAdapter' Endpoint for the + 'org.springframework.integration.ftp.inbound.FtpInboundStreamingMessageSource'. + + + + + + - - Specify a SpEL expression which - will be used to evaluate the directory - path from where the files will be transferred - (e.g., "@someBean.fetchDirectory"); - Mutually exclusive with 'remote-directory'. - + @@ -240,7 +196,7 @@ - + - + - + + + + + + + + + - Identifies the remote directory path (e.g., "/remote/mytransfers") - Mutually exclusive with 'remote-directory-expression'. + Identifies channel attached to this adapter. + The channel to which messages will be sent + by this adapter. @@ -537,7 +503,46 @@ use="optional"> - Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers") + Not used. + + + + + + + Allows you to provide a file name pattern to + determine the file names + that need to be scanned. + This is based on + simple pattern matching (e.g., "*.txt, fo*.txt" + etc.) + + + + + + + Allows you to provide a Regular Expression to + determine the file names + that need to be scanned. + (e.g., + "f[o]+\.txt" etc.) + + + + + + + + + + + + Allows you to specify a reference to a + [org.springframework.integration.file.filters.FileListFilter] + bean. This filter is applied to files on the remote server and + only files that pass the filter are retrieved. @@ -552,10 +557,24 @@ + + + + + + + Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers") + + + + + + + + - + @@ -569,17 +588,6 @@ ]]> - - - - Extension used when downloading files. We - change - it right after we know it's - downloaded. - - - - @@ -590,7 +598,39 @@ + + + + Identifies the remote directory path (e.g., "/remote/mytransfers") + Mutually exclusive with 'remote-directory-expression'. + + + + + + + Specify a SpEL expression which + will be used to evaluate the directory + path to where the files will be transferred + (e.g., "headers.['remote_dir'] + '/myTransfers'" for outbound endpoints) + There is no root object (message) for inbound endpoints + (e.g., "@someBean.fetchDirectory"); + + + + + + + + Extension used when downloading files. We change + it right after we know it's downloaded. + + + + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java new file mode 100644 index 0000000000..964d340e39 --- /dev/null +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java @@ -0,0 +1,151 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.ftp; + +import java.io.File; +import java.util.Arrays; + +import org.apache.commons.net.ftp.FTPFile; +import org.apache.ftpserver.FtpServer; +import org.apache.ftpserver.FtpServerFactory; +import org.apache.ftpserver.ftplet.Authentication; +import org.apache.ftpserver.ftplet.AuthenticationFailedException; +import org.apache.ftpserver.ftplet.FtpException; +import org.apache.ftpserver.ftplet.User; +import org.apache.ftpserver.ftplet.UserManager; +import org.apache.ftpserver.listener.Listener; +import org.apache.ftpserver.listener.ListenerFactory; +import org.apache.ftpserver.usermanager.impl.BaseUser; +import org.apache.ftpserver.usermanager.impl.ConcurrentLoginPermission; +import org.apache.ftpserver.usermanager.impl.TransferRatePermission; +import org.apache.ftpserver.usermanager.impl.WritePermission; +import org.junit.AfterClass; +import org.junit.BeforeClass; + +import org.springframework.integration.file.remote.RemoteFileTestSupport; +import org.springframework.integration.file.remote.session.CachingSessionFactory; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.ftp.session.DefaultFtpSessionFactory; + +/** + * Provides an embedded FTP Server for test cases. + * + * @author Artem Bilan + * @author Gary Russell + * @author David Turanski + * @since 4.2.12 + */ +public class FtpTestSupport extends RemoteFileTestSupport { + + private static volatile FtpServer server; + + public String getTargetLocalDirectoryName() { + return targetLocalDirectory.getAbsolutePath() + File.separator; + } + + @BeforeClass + public static void createServer() throws Exception { + FtpServerFactory serverFactory = new FtpServerFactory(); + serverFactory.setUserManager(new TestUserManager(remoteTemporaryFolder.getRoot().getAbsolutePath())); + + ListenerFactory factory = new ListenerFactory(); + factory.setPort(0); + serverFactory.addListener("default", factory.createListener()); + + server = serverFactory.createServer(); + server.start(); + + Listener listener = serverFactory.getListeners().values().iterator().next(); + port = listener.getPort(); + } + + + @AfterClass + public static void stopServer() throws Exception { + server.stop(); + } + + @Override + protected String prefix() { + return "ftp"; + } + + public static SessionFactory sessionFactory() { + DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory(); + sf.setHost("localhost"); + sf.setPort(port); + sf.setUsername("foo"); + sf.setPassword("foo"); + + return new CachingSessionFactory(sf); + } + + private static class TestUserManager implements UserManager { + + private final BaseUser testUser; + + private TestUserManager(String homeDirectory) { + this.testUser = new BaseUser(); + this.testUser.setAuthorities(Arrays.asList(new ConcurrentLoginPermission(1024, 1024), + new WritePermission(), + new TransferRatePermission(1024, 1024))); + this.testUser.setHomeDirectory(homeDirectory); + this.testUser.setName("TEST_USER"); + } + + + @Override + public User getUserByName(String s) throws FtpException { + return this.testUser; + } + + @Override + public String[] getAllUserNames() throws FtpException { + return new String[] { "TEST_USER" }; + } + + @Override + public void delete(String s) throws FtpException { + } + + @Override + public void save(User user) throws FtpException { + } + + @Override + public boolean doesExist(String s) throws FtpException { + return true; + } + + @Override + public User authenticate(Authentication authentication) throws AuthenticationFailedException { + return this.testUser; + } + + @Override + public String getAdminName() throws FtpException { + return "admin"; + } + + @Override + public boolean isAdmin(String s) throws FtpException { + return s.equals("admin"); + } + + } + +} diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests-context.xml new file mode 100644 index 0000000000..db90c2f0a2 --- /dev/null +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests-context.xml @@ -0,0 +1,48 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java new file mode 100644 index 0000000000..7040442bd7 --- /dev/null +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java @@ -0,0 +1,99 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.ftp.config; + +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.file.remote.session.CachingSessionFactory; +import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; +import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource; +import org.springframework.integration.ftp.session.DefaultFtpSessionFactory; +import org.springframework.integration.ftp.session.FtpSession; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class FtpStreamingInboundChannelAdapterParserTests { + + @Autowired + private SourcePollingChannelAdapter ftpInbound; + + @Autowired + private MessageChannel ftpChannel; + + @Autowired + private CachingSessionFactory csf; + + @Test + public void testFtpInboundChannelAdapterComplete() throws Exception { + assertFalse(TestUtils.getPropertyValue(this.ftpInbound, "autoStartup", Boolean.class)); + assertEquals("ftpInbound", this.ftpInbound.getComponentName()); + assertEquals("ftp:inbound-streaming-channel-adapter", this.ftpInbound.getComponentType()); + assertSame(this.ftpChannel, TestUtils.getPropertyValue(this.ftpInbound, "outputChannel")); + FtpStreamingMessageSource source = TestUtils.getPropertyValue(ftpInbound, "source", + FtpStreamingMessageSource.class); + + assertNotNull(TestUtils.getPropertyValue(source, "comparator")); + assertThat(TestUtils.getPropertyValue(source, "remoteFileSeparator", String.class), equalTo("X")); + assertThat(TestUtils.getPropertyValue(source, "filter"), instanceOf(FtpSimplePatternFileListFilter.class)); + assertSame(this.csf, TestUtils.getPropertyValue(source, "remoteFileTemplate.sessionFactory")); + } + + public static class TestSessionFactoryBean implements FactoryBean { + + @Override + public DefaultFtpSessionFactory getObject() throws Exception { + DefaultFtpSessionFactory factory = mock(DefaultFtpSessionFactory.class); + FtpSession session = mock(FtpSession.class); + when(factory.getSession()).thenReturn(session); + return factory; + } + + @Override + public Class getObjectType() { + return DefaultFtpSessionFactory.class; + } + + @Override + public boolean isSingleton() { + return true; + } + } + +} diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java new file mode 100644 index 0000000000..06675f3e76 --- /dev/null +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java @@ -0,0 +1,128 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.ftp.inbound; + +import static org.hamcrest.Matchers.equalTo; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; + +import java.io.InputStream; + +import org.apache.commons.net.ftp.FTPFile; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.Transformer; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.ftp.FtpTestSupport; +import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter; +import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; +import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.transformer.StreamTransformer; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.scheduling.support.PeriodicTrigger; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.2.12 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class FtpStreamingMessageSourceTests extends FtpTestSupport { + + @Autowired + public PollableChannel data; + + @SuppressWarnings("unchecked") + @Test + public void testAllContents() { + Message received = (Message) this.data.receive(10000); + assertNotNull(received); + assertThat(new String(received.getPayload()), equalTo("source1")); + received = (Message) this.data.receive(10000); + assertNotNull(received); + assertThat(new String(received.getPayload()), equalTo("source2")); + assertNull(this.data.receive(0)); + } + + @Configuration + @EnableIntegration + public static class Config { + + @Bean + public QueueChannel data() { + return new QueueChannel(); + } + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata defaultPoller() { + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(new PeriodicTrigger(500)); + pollerMetadata.setMaxMessagesPerPoll(2000); + return pollerMetadata; + } + + @Bean + public SubscribableChannel stream() { + return new DirectChannel(); + } + + @Bean + @InboundChannelAdapter("stream") + public MessageSource ftpMessageSource() { + FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null); + messageSource.setRemoteDirectory("ftpSource/"); + messageSource.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "foo")); + return messageSource; + } + + @Bean + @Transformer(inputChannel = "stream", outputChannel = "data") + public org.springframework.integration.transformer.Transformer transformer() { + return new StreamTransformer(); + } + + @Bean + public FtpRemoteFileTemplate template() { + return new FtpRemoteFileTemplate(ftpSessionFactory()); + } + + @Bean + public SessionFactory ftpSessionFactory() { + return FtpStreamingMessageSourceTests.sessionFactory(); + } + + } + +} diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests-context.xml deleted file mode 100644 index 4a7e5793fb..0000000000 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests-context.xml +++ /dev/null @@ -1,15 +0,0 @@ - - - - - - - - diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java index aaef98ba8c..76d6a296d9 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java @@ -29,12 +29,12 @@ import java.util.UUID; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; -import org.junit.After; -import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.remote.ClientCallbackWithoutResult; @@ -42,7 +42,7 @@ import org.springframework.integration.file.remote.SessionCallback; import org.springframework.integration.file.remote.SessionCallbackWithoutResult; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; -import org.springframework.integration.ftp.TestFtpServer; +import org.springframework.integration.ftp.FtpTestSupport; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.ContextConfiguration; @@ -55,21 +55,11 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) -public class FtpRemoteFileTemplateTests { - - @Autowired - private TestFtpServer ftpServer; +public class FtpRemoteFileTemplateTests extends FtpTestSupport { @Autowired private SessionFactory sessionFactory; - @Before - @After - public void setup() { - this.ftpServer.recursiveDelete(ftpServer.getTargetLocalDirectory()); - this.ftpServer.recursiveDelete(ftpServer.getTargetFtpDirectory()); - } - @Test public void testINT3412AppendStatRmdir() throws IOException { FtpRemoteFileTemplate template = new FtpRemoteFileTemplate(sessionFactory); @@ -142,4 +132,14 @@ public class FtpRemoteFileTemplateTests { newFile.delete(); } + @Configuration + public static class Config { + + @Bean + public SessionFactory ftpSessionFactory() { + return FtpRemoteFileTemplateTests.sessionFactory(); + } + + } + } 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 index 39874bcefe..ae9a6bf708 100644 --- 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 @@ -17,6 +17,8 @@ package org.springframework.integration.sftp.config; import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer; import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter; import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter; import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer; @@ -37,18 +39,18 @@ public class SftpInboundChannelAdapterParser extends AbstractRemoteFileInboundCh } @Override - protected String getInboundFileSynchronizerClassname() { - return SftpInboundFileSynchronizer.class.getName(); + protected Class getInboundFileSynchronizerClass() { + return SftpInboundFileSynchronizer.class; } @Override - protected String getSimplePatternFileListFilterClassname() { - return SftpSimplePatternFileListFilter.class.getName(); + protected Class> getSimplePatternFileListFilterClass() { + return SftpSimplePatternFileListFilter.class; } @Override - protected String getRegexPatternFileListFilterClassname() { - return SftpRegexPatternFileListFilter.class.getName(); + protected Class> getRegexPatternFileListFilterClass() { + return SftpRegexPatternFileListFilter.class; } } 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 46338fd4cf..fe8077d5c8 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 @@ -32,6 +32,7 @@ public class SftpNamespaceHandler extends AbstractIntegrationNamespaceHandler { @Override public void init() { registerBeanDefinitionParser("inbound-channel-adapter", new SftpInboundChannelAdapterParser()); + registerBeanDefinitionParser("inbound-streaming-channel-adapter", new SftpStreamingInboundChannelAdapterParser()); registerBeanDefinitionParser("outbound-channel-adapter", new SftpOutboundChannelAdapterParser()); registerBeanDefinitionParser("outbound-gateway", new SftpOutboundGatewayParser()); } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java new file mode 100644 index 0000000000..5baf4c6d58 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java @@ -0,0 +1,55 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.sftp.config; + +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.file.config.AbstractRemoteFileStreamingInboundChannelAdapterParser; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.RemoteFileOperations; +import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter; +import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter; +import org.springframework.integration.sftp.inbound.SftpStreamingMessageSource; +import org.springframework.integration.sftp.session.SftpRemoteFileTemplate; + +/** + * @author Gary Russell + * @since 4.2.12 + * + */ +public class SftpStreamingInboundChannelAdapterParser extends AbstractRemoteFileStreamingInboundChannelAdapterParser { + + @Override + protected Class> getTemplateClass() { + return SftpRemoteFileTemplate.class; + } + + @Override + protected Class> getMessageSourceClass() { + return SftpStreamingMessageSource.class; + } + + @Override + protected Class> getSimplePatternFileListFilterClass() { + return SftpSimplePatternFileListFilter.class; + } + + @Override + protected Class> getRegexPatternFileListFilterClass() { + return SftpRegexPatternFileListFilter.class; + } + +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java new file mode 100644 index 0000000000..e66ec79a4c --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java @@ -0,0 +1,74 @@ +/* + * Copyright 2016 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.inbound; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Comparator; +import java.util.List; + +import org.springframework.integration.file.remote.AbstractFileInfo; +import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.sftp.session.SftpFileInfo; + +import com.jcraft.jsch.ChannelSftp.LsEntry; + +/** + * Message source for streaming SFTP remote file contents. + * + * @author Gary Russell + * @since 4.2.12 + * + */ +public class SftpStreamingMessageSource extends AbstractRemoteFileStreamingMessageSource { + + /** + * Construct an instance with the supplied template. + * @param template the template. + */ + public SftpStreamingMessageSource(RemoteFileTemplate template) { + super(template, null); + } + + /** + * Construct an instance with the supplied template and comparator. + * Note: the comparator is applied each time the remote directory is listed + * which only occurs when the previous list is exhausted. + * @param template the template. + * @param comparator the comparator. + */ + public SftpStreamingMessageSource(RemoteFileTemplate template, + Comparator> comparator) { + super(template, comparator); + } + + @Override + public String getComponentType() { + return "sftp:inbound-streaming-channel-adapter"; + } + + @Override + protected List> asFileInfoList(Collection files) { + List> canonicalFiles = new ArrayList>(); + for (LsEntry file : files) { + canonicalFiles.add(new SftpFileInfo(file)); + } + return canonicalFiles; + } + +} diff --git a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.2.xsd b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.2.xsd index 33d6efcd8e..6ff4aad577 100644 --- a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.2.xsd +++ b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.2.xsd @@ -22,7 +22,7 @@ - + - + + Configures a 'SourcePollingChannelAdapter' Endpoint for the + 'org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource' + that synchronizes with a remote SFTP endpoint. + - - - - - - - - - - - - - Identifies channel attached to this adapter. This channel where messages will be sent - to by this adapter. - - - - - - - - - - - - - - - - - Allows you to specify a reference to a - [org.springframework.integration.file.filters.FileListFilter] - bean. This filter is applied to files on the remote server and - only files that pass the filter are retrieved. - - - - - - - Allows you to provide a file name pattern to - determine the file names - that need to be scanned. - This is based on - simple pattern matching (e.g., "*.txt, fo*.txt" - etc.) - - - - + + Allows you to provide a SpEL expression to @@ -148,15 +96,12 @@ - + - - Allows you to provide a Regular Expression to - determine the file names - that need to be scanned. - (e.g., - "f[o]+\.txt" etc.) - + @@ -219,16 +164,29 @@ - + + + + + + + + + + Configures a 'SourcePollingChannelAdapter' Endpoint for the + 'org.springframework.integration.ftp.inbound.FtpInboundStreamingMessageSource'. + + + + + + - - Specify a SpEL expression which - will be used to evaluate the directory - path from where the files will be transferred - (e.g., "@someBean.fetchDirectory"). - Mutually exclusive with 'remote-directory'. - + @@ -244,7 +202,7 @@ - + - + Tells this adapter if local directory must be @@ -524,15 +481,63 @@ - + - + + + + + + + + + + + + Identifies channel attached to this adapter. + The channel to which messages will be sent + by this adapter. + + + + - Identifies the directory path (e.g., - "/temp/mytransfers") - Mutually exclusive with 'remote-directory-expression'. + Allows you to provide a file name pattern to + determine the file names + that need to be scanned. + This is based on + simple pattern matching (e.g., "*.txt, fo*.txt" + etc.) + + + + + + + Allows you to provide a Regular Expression to + determine the file names + that need to be scanned. + (e.g., + "f[o]+\.txt" etc.) + + + + + + + + + + + + Allows you to specify a reference to a + [org.springframework.integration.file.filters.FileListFilter] + bean. This filter is applied to files on the remote server and + only files that pass the filter are retrieved. @@ -540,7 +545,7 @@ use="optional"> - Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers") + Not used. @@ -557,10 +562,24 @@ + + + + + + + Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers") + + + + + + + + - + @@ -574,16 +593,6 @@ ]]> - - - - Extension used when downloading files. We - change - it right after we know it's - downloaded. - - - @@ -594,7 +603,39 @@ + + + + Identifies the remote directory path (e.g., "/remote/mytransfers") + Mutually exclusive with 'remote-directory-expression'. + + + + + + + Specify a SpEL expression which + will be used to evaluate the directory + path to where the files will be transferred + (e.g., "headers.['remote_dir'] + '/myTransfers'" for outbound endpoints) + There is no root object (message) for inbound endpoints + (e.g., "@someBean.fetchDirectory"); + + + + + + + + Extension used when downloading files. We change + it right after we know it's downloaded. + + + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java new file mode 100644 index 0000000000..32f7b1a650 --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java @@ -0,0 +1,97 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.sftp; + +import java.io.File; +import java.util.Collections; + +import org.apache.sshd.SshServer; +import org.apache.sshd.common.NamedFactory; +import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory; +import org.apache.sshd.server.Command; +import org.apache.sshd.server.PasswordAuthenticator; +import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider; +import org.apache.sshd.server.sftp.SftpSubsystem; +import org.junit.AfterClass; +import org.junit.BeforeClass; + +import org.springframework.integration.file.remote.RemoteFileTestSupport; +import org.springframework.integration.file.remote.session.CachingSessionFactory; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; + +import com.jcraft.jsch.ChannelSftp.LsEntry; + +/** + * Provides an embedded SFTP Server for test cases. + * + * @author David Turanski + * @author Gary Russell + * @since 4.2.12 + */ +public class SftpTestSupport extends RemoteFileTestSupport { + + private static SshServer server; + + public String getTargetLocalDirectoryName() { + return targetLocalDirectory.getAbsolutePath() + File.separator; + } + + @Override + public String prefix() { + return "sftp"; + } + + @BeforeClass + public static void createServer() throws Exception { + server = SshServer.setUpDefaultServer(); + server.setPasswordAuthenticator(new PasswordAuthenticator() { + + @Override + public boolean authenticate(String username, String password, + org.apache.sshd.server.session.ServerSession session) { + return true; + } + }); + server.setPort(0); + server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser")); + server.setSubsystemFactories(Collections.>singletonList(new SftpSubsystem.Factory())); + server.setFileSystemFactory(new VirtualFileSystemFactory(remoteTemporaryFolder.getRoot().getAbsolutePath())); + server.start(); + port = server.getPort(); + } + + public static SessionFactory sessionFactory() { + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true); + factory.setHost("localhost"); + factory.setPort(port); + factory.setUser("foo"); + factory.setPassword("foo"); + factory.setAllowUnknownKeys(true); + return new CachingSessionFactory(factory); + } + + @AfterClass + public static void stopServer() throws Exception { + server.stop(); + File hostkey = new File("hostkey.ser"); + if (hostkey.exists()) { + hostkey.delete(); + } + } + +} diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests-context.xml new file mode 100644 index 0000000000..407723092c --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests-context.xml @@ -0,0 +1,48 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java new file mode 100644 index 0000000000..bc992f342f --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java @@ -0,0 +1,99 @@ +/* + * Copyright 2016 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 static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.file.remote.session.CachingSessionFactory; +import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter; +import org.springframework.integration.sftp.inbound.SftpStreamingMessageSource; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; +import org.springframework.integration.sftp.session.SftpSession; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class SftpStreamingInboundChannelAdapterParserTests { + + @Autowired + private SourcePollingChannelAdapter sftpInbound; + + @Autowired + private MessageChannel sftpChannel; + + @Autowired + private CachingSessionFactory csf; + + @Test + public void testFtpInboundChannelAdapterComplete() throws Exception { + assertFalse(TestUtils.getPropertyValue(this.sftpInbound, "autoStartup", Boolean.class)); + assertEquals("sftpInbound", this.sftpInbound.getComponentName()); + assertEquals("sftp:inbound-streaming-channel-adapter", this.sftpInbound.getComponentType()); + assertSame(this.sftpChannel, TestUtils.getPropertyValue(this.sftpInbound, "outputChannel")); + SftpStreamingMessageSource source = TestUtils.getPropertyValue(sftpInbound, "source", + SftpStreamingMessageSource.class); + + assertNotNull(TestUtils.getPropertyValue(source, "comparator")); + assertThat(TestUtils.getPropertyValue(source, "remoteFileSeparator", String.class), equalTo("X")); + assertThat(TestUtils.getPropertyValue(source, "filter"), instanceOf(SftpSimplePatternFileListFilter.class)); + assertSame(this.csf, TestUtils.getPropertyValue(source, "remoteFileTemplate.sessionFactory")); + } + + public static class TestSessionFactoryBean implements FactoryBean { + + @Override + public DefaultSftpSessionFactory getObject() throws Exception { + DefaultSftpSessionFactory factory = mock(DefaultSftpSessionFactory.class); + SftpSession session = mock(SftpSession.class); + when(factory.getSession()).thenReturn(session); + return factory; + } + + @Override + public Class getObjectType() { + return DefaultSftpSessionFactory.class; + } + + @Override + public boolean isSingleton() { + return true; + } + } + +} diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml index 727895111f..2f6ab0bc2f 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml @@ -7,13 +7,15 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd"> + + @@ -34,6 +36,4 @@ - - diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java index 58ab0781b6..965c389bb4 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -30,10 +30,15 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.sftp.SftpTestSupport; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import com.jcraft.jsch.ChannelSftp.LsEntry; + /** * @author Gary Russell * @author Artem Bilan @@ -43,12 +48,12 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext -public class RollbackLocalFilterTests { +public class RollbackLocalFilterTests extends SftpTestSupport { @BeforeClass @AfterClass public static void clean() { - new File("local-test-dir/rollback/sftpSource1.txt").delete(); + new File("local-test-dir/rollback/sftpSource2.txt").delete(); } @Autowired @@ -57,7 +62,7 @@ public class RollbackLocalFilterTests { @Test public void testRollback() throws Exception { assertTrue(this.crash.getLatch().await(10, TimeUnit.SECONDS)); - assertEquals("sftpSource1.txt", this.crash.getFile().getName()); + assertEquals("sftpSource2.txt", this.crash.getFile().getName()); } public static class Crash { @@ -86,4 +91,13 @@ public class RollbackLocalFilterTests { } } + public static class Config { + + @Bean + public SessionFactory sftpSessionFactory() { + return RollbackLocalFilterTests.sessionFactory(); + } + + } + } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java new file mode 100644 index 0000000000..f4ced9d76f --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java @@ -0,0 +1,128 @@ +/* + * Copyright 2016 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.inbound; + +import static org.hamcrest.Matchers.equalTo; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; + +import java.io.InputStream; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.Transformer; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.file.filters.AcceptOnceFileListFilter; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.sftp.SftpTestSupport; +import org.springframework.integration.sftp.session.SftpRemoteFileTemplate; +import org.springframework.integration.transformer.StreamTransformer; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.scheduling.support.PeriodicTrigger; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.jcraft.jsch.ChannelSftp.LsEntry; + +/** + * @author Gary Russell + * @since 4.2.12 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class SftpStreamingMessageSourceTests extends SftpTestSupport { + + @Autowired + public PollableChannel data; + + @SuppressWarnings("unchecked") + @Test + public void testAllContents() { + Message received = (Message) this.data.receive(10000); + assertNotNull(received); + assertThat(new String(received.getPayload()), equalTo("source1")); + received = (Message) this.data.receive(10000); + assertNotNull(received); + assertThat(new String(received.getPayload()), equalTo("source2")); + assertNull(this.data.receive(0)); + } + + @Configuration + @EnableIntegration + public static class Config { + + @Bean + public QueueChannel data() { + return new QueueChannel(); + } + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata defaultPoller() { + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(new PeriodicTrigger(500)); + pollerMetadata.setMaxMessagesPerPoll(2000); + return pollerMetadata; + } + + @Bean + @InboundChannelAdapter("stream") + public MessageSource ftpMessageSource() { + SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), null); + messageSource.setRemoteDirectory("sftpSource/"); + messageSource.setFilter(new AcceptOnceFileListFilter()); + return messageSource; + } + + @Bean + public SubscribableChannel stream() { + return new DirectChannel(); + } + + @Bean + @Transformer(inputChannel = "stream", outputChannel = "data") + public org.springframework.integration.transformer.Transformer transformer() { + return new StreamTransformer(); + } + + @Bean + public SftpRemoteFileTemplate template() { + return new SftpRemoteFileTemplate(ftpSessionFactory()); + } + + @Bean + public SessionFactory ftpSessionFactory() { + return SftpStreamingMessageSourceTests.sessionFactory(); + } + + } + +} diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java index 6279d1fb1c..84d6776845 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java @@ -21,19 +21,20 @@ import static org.junit.Assert.assertTrue; import java.io.IOException; -import org.junit.After; -import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.remote.ClientCallbackWithoutResult; import org.springframework.integration.file.remote.SessionCallback; import org.springframework.integration.file.remote.SessionCallbackWithoutResult; import org.springframework.integration.file.remote.session.Session; -import org.springframework.integration.sftp.TestSftpServer; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.sftp.SftpTestSupport; import org.springframework.integration.sftp.TestSftpServerConfig; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; @@ -53,21 +54,11 @@ import com.jcraft.jsch.SftpException; @ContextConfiguration(classes=TestSftpServerConfig.class) @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext -public class SftpRemoteFileTemplateTests { - - @Autowired - private TestSftpServer sftpServer; +public class SftpRemoteFileTemplateTests extends SftpTestSupport { @Autowired private DefaultSftpSessionFactory sessionFactory; - @Before - @After - public void setup() { - this.sftpServer.recursiveDelete(sftpServer.getTargetLocalDirectory()); - this.sftpServer.recursiveDelete(sftpServer.getTargetSftpDirectory()); - } - @Test public void testINT3412AppendStatRmdir() { SftpRemoteFileTemplate template = new SftpRemoteFileTemplate(sessionFactory); @@ -115,4 +106,14 @@ public class SftpRemoteFileTemplateTests { assertFalse(template.exists("foo")); } + @Configuration + public static class Config { + + @Bean + public SessionFactory ftpSessionFactory() { + return SftpRemoteFileTemplateTests.sessionFactory(); + } + + } + }