From 287d924fc0024dab2833d6c24de8d10d770c8c90 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 27 Apr 2016 15:10:02 -0400 Subject: [PATCH] INT-4015: Streaming Remote File Inbound Adapter JIRA: https://jira.spring.io/browse/INT-4015 https://jira.spring.io/browse/INT-3854 Initial commit. Reworked to emit an input stream and use the file splitter. Add StreamTransformer. Add CLOSABLE_RESOURCE header so we can close the session automatically. Implement INT-3854, FTP, SFTP (S)FTP Namespace Changes Docs - also fixes a PDF overflow Polishing - PR Comments checkstyle fixes Polishing - Add Namespace for StreamParser Polishing - PR Comments --- .../IntegrationMessageHeaderAccessor.java | 16 ++ .../xml/IntegrationNamespaceHandler.java | 1 + .../config/xml/StreamTransformerParser.java | 35 +++ .../transformer/StreamTransformer.java | 70 +++++ .../config/spring-integration-4.3.xsd | 47 +++- .../ObjectToStringTransformerParserTests.java | 10 +- .../StreamTransformerParserTests-context.xml | 30 ++ .../xml/StreamTransformerParserTests.java | 88 ++++++ .../SyslogTransformerParserTests-context.xml | 9 + .../integration/file/FileHeaders.java | 4 + ...RemoteFileInboundChannelAdapterParser.java | 54 +--- ...eStreamingInboundChannelAdapterParser.java | 76 +++++ .../file/config/FileParserUtils.java | 40 +++ ...tractRemoteFileStreamingMessageSource.java | 187 +++++++++++++ .../file/remote/RemoteFileOperations.java | 16 ++ .../file/remote/RemoteFileTemplate.java | 18 ++ .../AbstractRemoteFileOutboundGateway.java | 7 +- .../file/remote/session/Session.java | 6 +- .../file/splitter/FileSplitter.java | 20 +- .../config/spring-integration-file-4.3.xsd | 10 +- .../file/remote/RemoteFileTestSupport.java | 177 ++++++++++++ .../file/remote/StreamingInboundTests.java | 221 +++++++++++++++ .../FtpInboundChannelAdapterParser.java | 14 +- .../ftp/config/FtpNamespaceHandler.java | 2 + ...pStreamingInboundChannelAdapterParser.java | 55 ++++ .../inbound/FtpStreamingMessageSource.java | 74 +++++ .../integration/ftp/session/FtpSession.java | 6 +- .../ftp/config/spring-integration-ftp-4.3.xsd | 209 +++++++------- .../integration/ftp/FtpTestSupport.java | 151 ++++++++++ .../integration/ftp/TestFtpServer.java | 260 ------------------ ...boundChannelAdapterParserTests-context.xml | 2 - ...boundChannelAdapterParserTests-context.xml | 48 ++++ ...amingInboundChannelAdapterParserTests.java | 99 +++++++ ...boundChannelAdapterParserTests-context.xml | 14 +- .../FtpStreamingMessageSourceTests.java | 116 ++++++++ .../FtpServerOutboundTests-context.xml | 12 +- .../ftp/outbound/FtpServerOutboundTests.java | 48 +++- .../FtpRemoteFileTemplateTests-context.xml | 15 - .../session/FtpRemoteFileTemplateTests.java | 28 +- .../rollback/sftpSource2.txt.writing | 0 .../SftpInboundChannelAdapterParser.java | 14 +- .../sftp/config/SftpNamespaceHandler.java | 1 + ...pStreamingInboundChannelAdapterParser.java | 55 ++++ .../inbound/SftpStreamingMessageSource.java | 74 +++++ .../config/spring-integration-sftp-4.3.xsd | 226 ++++++++------- .../integration/sftp/SftpTestSupport.java | 97 +++++++ .../integration/sftp/TestSftpServer.java | 186 ------------- .../sftp/TestSftpServerConfig.java | 43 --- ...boundChannelAdapterParserTests-context.xml | 48 ++++ ...amingInboundChannelAdapterParserTests.java | 99 +++++++ .../RollbackLocalFilterTests-context.xml | 6 +- .../inbound/RollbackLocalFilterTests.java | 22 +- .../SftpStreamingMessageSourceTests.java | 117 ++++++++ .../SftpServerOutboundTests-context.xml | 12 +- .../outbound/SftpServerOutboundTests.java | 58 ++-- .../session/SftpRemoteFileTemplateTests.java | 32 +-- src/reference/asciidoc/file.adoc | 10 +- src/reference/asciidoc/ftp.adoc | 80 ++++-- src/reference/asciidoc/sftp.adoc | 89 ++++-- src/reference/asciidoc/transformer.adoc | 37 ++- src/reference/asciidoc/whats-new.adoc | 20 ++ 61 files changed, 2703 insertions(+), 918 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/config/xml/StreamTransformerParser.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java create mode 100644 spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java create mode 100644 spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java create mode 100644 spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java delete mode 100644 spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TestFtpServer.java create mode 100644 spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests-context.xml create mode 100644 spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java create mode 100644 spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java delete mode 100644 spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests-context.xml create mode 100644 spring-integration-sftp/local-test-dir/rollback/sftpSource2.txt.writing create mode 100644 spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java create mode 100644 spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java create mode 100644 spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java delete mode 100644 spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSftpServer.java delete mode 100644 spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSftpServerConfig.java create mode 100644 spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests-context.xml create mode 100644 spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java create mode 100644 spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java index 5247c203ea..d9fe66cc00 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java @@ -16,6 +16,7 @@ package org.springframework.integration; +import java.io.Closeable; import java.util.Date; import java.util.Map; @@ -50,6 +51,8 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { public static final String DUPLICATE_MESSAGE = "duplicateMessage"; + public static final String CLOSEABLE_RESOURCE = "closableResource"; + public IntegrationMessageHeaderAccessor(Message message) { super(message); } @@ -76,6 +79,19 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { return this.getHeader(PRIORITY, Integer.class); } + /** + * If the payload was created by a {@link Closeable} that needs to remain + * open until the payload is consumed, the resource will be added to this + * header. After the payload is consumed the {@link Closeable} should be + * closed. Usually this must occur in an endpoint close to the message + * origin in the flow, and in the same JVM. + * @return the {@link Closeable}. + * @since 4.3 + */ + public Closeable getCloseableResource() { + return this.getHeader(CLOSEABLE_RESOURCE, Closeable.class); + } + @SuppressWarnings("unchecked") public T getHeader(String key, Class type) { Object value = getHeader(key); 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..c2e2df51df --- /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.3 + * + */ +public class StreamTransformerParser extends ObjectToStringTransformerParser { + + @Override + protected String getTransformerClassName() { + return StreamTransformer.class.getName(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java new file mode 100644 index 0000000000..24196313ea --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java @@ -0,0 +1,70 @@ +/* + * 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.transformer; + +import java.io.ByteArrayOutputStream; +import java.io.Closeable; +import java.io.InputStream; + +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.util.FileCopyUtils; + +/** + * Transforms an InputStream payload to a byte[] or String (if a + * charset is provided). + * + * @author Gary Russell + * @since 4.3 + * + */ +public class StreamTransformer extends AbstractTransformer { + + private final String charset; + + /** + * Construct an instance to transform an {@link InputStream} to + * a {@code byte[]}. + */ + public StreamTransformer() { + this(null); + } + + /** + * Construct an instance with the charset to convert the stream to a + * String; if null a {@code byte[]} will be produced instead. + * @param charset the charset. + */ + public StreamTransformer(String charset) { + this.charset = charset; + } + + @Override + protected Object doTransform(Message message) throws Exception { + Assert.isTrue(message.getPayload() instanceof InputStream, "payload must be an InputStream"); + InputStream stream = (InputStream) message.getPayload(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(stream, baos); + Closeable closeableResource = new IntegrationMessageHeaderAccessor(message).getCloseableResource(); + if (closeableResource != null) { + closeableResource.close(); + } + return this.charset == null ? baos.toByteArray() : baos.toString(this.charset); + } + +} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd index d3296c7ba6..d3635ff66b 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd @@ -1826,6 +1826,8 @@ + + @@ -2686,11 +2688,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[]. + + + + + + + @@ -2789,8 +2829,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/FileHeaders.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java index b9e942a742..faa0e6b2f9 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java @@ -35,6 +35,10 @@ public abstract class FileHeaders { public static final String REMOTE_FILE = PREFIX + "remoteFile"; + /** + * @deprecated - use {@code IntegrationMessageHeaderAccessor#CLOSEABLE_RESOURCE}. + */ + @Deprecated public static final String REMOTE_SESSION = PREFIX + "remoteSession"; public static final String RENAME_TO = PREFIX + "renameTo"; 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 3c4a367c78..b822ee7dfa 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; /** @@ -41,7 +43,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")); @@ -57,7 +59,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 = @@ -82,53 +85,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..ca02be1eb7 --- /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.3 + */ +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 fb99d04cdc..4699999bcf 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 @@ -23,6 +23,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; @@ -89,4 +90,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 new file mode 100644 index 0000000000..2303178534 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java @@ -0,0 +1,187 @@ +/* + * 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.remote; + +import java.io.IOException; +import java.io.InputStream; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.endpoint.AbstractMessageSource; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.session.Session; +import org.springframework.messaging.MessagingException; +import org.springframework.util.Assert; + +/** + * A message source that produces a message with an {@link InputStream} payload + * referencing a remote file. + * + * @author Gary Russell + * @since 4.3 + * + */ +public abstract class AbstractRemoteFileStreamingMessageSource extends AbstractMessageSource + implements BeanFactoryAware, InitializingBean { + + private final RemoteFileTemplate remoteFileTemplate; + + private final BlockingQueue> toBeReceived = new LinkedBlockingQueue>(); + + private final Comparator> comparator; + + /** + * the path on the remote server. + */ + private volatile Expression remoteDirectoryExpression; + + private volatile String remoteFileSeparator = "/"; + + /** + * An {@link FileListFilter} that runs against the remote file system view. + */ + private volatile FileListFilter filter; + + protected AbstractRemoteFileStreamingMessageSource(RemoteFileTemplate template, + Comparator> comparator) { + this.remoteFileTemplate = template; + this.comparator = comparator; + } + + /** + * Specify the full path to the remote directory. + * + * @param remoteDirectory The remote directory. + */ + public void setRemoteDirectory(String remoteDirectory) { + this.remoteDirectoryExpression = new LiteralExpression(remoteDirectory); + } + + /** + * Specify an expression that evaluates to the full path to the remote directory. + * + * @param remoteDirectoryExpression The remote directory expression. + */ + public void setRemoteDirectoryExpression(Expression remoteDirectoryExpression) { + Assert.notNull(remoteDirectoryExpression, "'remoteDirectoryExpression' must not be null"); + this.remoteDirectoryExpression = remoteDirectoryExpression; + } + + /** + * Set the remote file separator; default '/' + * @param remoteFileSeparator the remote file separator. + */ + public void setRemoteFileSeparator(String remoteFileSeparator) { + Assert.notNull(remoteFileSeparator, "'remoteFileSeparator' must not be null"); + this.remoteFileSeparator = remoteFileSeparator; + } + + /** + * Set the filter to be applied to the remote files before transferring. + * @param filter the file list filter. + */ + public void setFilter(FileListFilter filter) { + this.filter = filter; + } + + protected RemoteFileTemplate getRemoteFileTemplate() { + return this.remoteFileTemplate; + } + + @Override + public final void afterPropertiesSet() { + Assert.state(this.remoteDirectoryExpression != null, "'remoteDirectoryExpression' must not be null"); + doInit(); + } + + /** + * Subclasses can override to perform initialization - called from + * {@link InitializingBean#afterPropertiesSet()}. + */ + protected void doInit() { + } + + @Override + protected Object doReceive() { + AbstractFileInfo file = poll(); + if (file != null) { + String remotePath = remotePath(file); + Session session = this.remoteFileTemplate.getSesssion(); + try { + return getMessageBuilderFactory().withPayload(session.readRaw(remotePath)) + .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session) + .setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory()) + .setHeader(FileHeaders.REMOTE_FILE, file.getFilename()) + .build(); + } + catch (IOException e) { + return new MessagingException("IOException when retrieving " + remotePath, e); + } + } + return null; + } + + protected AbstractFileInfo poll() { + if (this.toBeReceived.size() == 0) { + listFiles(); + } + return this.toBeReceived.poll(); + } + + protected String remotePath(AbstractFileInfo file) { + String remotePath = file.getRemoteDirectory().endsWith(this.remoteFileSeparator) + ? file.getRemoteDirectory() + file.getFilename() + : file.getRemoteDirectory() + this.remoteFileSeparator + file.getFilename(); + return remotePath; + } + + private void listFiles() { + String remoteDirectory = this.remoteDirectoryExpression.getValue(getEvaluationContext(), String.class); + F[] files = this.remoteFileTemplate.list(remoteDirectory); + List filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files); + List> fileInfoList = asFileInfoList(filteredFiles); + Iterator> iterator = fileInfoList.iterator(); + while (iterator.hasNext()) { + AbstractFileInfo next = iterator.next(); + if (next.isDirectory()) { + iterator.remove(); + } + else { + next.setRemoteDirectory(remoteDirectory); + } + } + if (this.comparator != null) { + Collections.sort(fileInfoList, this.comparator); + } + this.toBeReceived.addAll(fileInfoList); + } + + abstract protected List> asFileInfoList(Collection files); + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java index 4219a31bea..f2343a99ab 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java @@ -16,6 +16,7 @@ package org.springframework.integration.file.remote; +import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.support.FileExistsMode; import org.springframework.messaging.Message; @@ -117,6 +118,13 @@ public interface RemoteFileOperations { */ void rename(String fromPath, String toPath); + /** + * List the files at the remote path. + * @param path the path. + * @return the list. + */ + F[] list(String path); + /** * Execute the callback's doInSession method after obtaining a session. * Reliably closes the session when the method exits. @@ -141,4 +149,12 @@ public interface RemoteFileOperations { */ T executeWithClient(ClientCallback callback); + /** + * Obtain a raw Session object. User must close the session when it is no longer + * needed. + * @return a session. + * @since 4.3 + */ + Session getSesssion(); + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java index 2909bb20e3..4f0f55a909 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java @@ -401,6 +401,24 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ }); } + + @Override + public F[] list(final String path) { + return this.execute(new SessionCallback() { + + @Override + public F[] doInSession(Session session) throws IOException { + return session.list(path); + } + + }); + } + + @Override + public Session getSesssion() { + return this.sessionFactory.getSession(); + } + @SuppressWarnings("rawtypes") @Override public T execute(SessionCallback callback) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index 9a50c5cf45..25217c4ccc 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -35,6 +35,7 @@ import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.filters.FileListFilter; @@ -586,7 +587,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply return getMessageBuilderFactory().withPayload(payload) .setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir) .setHeader(FileHeaders.REMOTE_FILE, remoteFilename) - .setHeader(FileHeaders.REMOTE_SESSION, session) + .setHeader("file_remoteSession", session) // TODO: remove in 5.0 + .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session) .build(); } @@ -1013,6 +1015,9 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply else if (e instanceof IOException) { throw (IOException) e; } + else { + throw new MessagingException("Failed to process MGET on first file", e); + } } return files; } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java index 4549c2e7b1..9aa8a7a3f3 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 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. @@ -16,6 +16,7 @@ package org.springframework.integration.file.remote.session; +import java.io.Closeable; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -30,7 +31,7 @@ import java.io.OutputStream; * @author Gary Russell * @since 2.0 */ -public interface Session { +public interface Session extends Closeable { boolean remove(String path) throws IOException; @@ -62,6 +63,7 @@ public interface Session { void rename(String pathFrom, String pathTo) throws IOException; + @Override void close(); boolean isOpen(); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java index 727074be80..84131da79e 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java @@ -17,6 +17,7 @@ package org.springframework.integration.file.splitter; import java.io.BufferedReader; +import java.io.Closeable; import java.io.File; import java.io.FileInputStream; import java.io.FileNotFoundException; @@ -33,6 +34,7 @@ import java.util.List; import java.util.Map; import java.util.NoSuchElementException; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.splitter.FileSplitter.FileMarker.Mark; import org.springframework.integration.splitter.AbstractMessageSplitter; @@ -158,7 +160,23 @@ public class FileSplitter extends AbstractMessageSplitter { return message; } - final BufferedReader bufferedReader = new BufferedReader(reader); + final BufferedReader bufferedReader = new BufferedReader(reader) { + + @Override + public void close() throws IOException { + try { + super.close(); + } + finally { + Closeable closeableResource = new IntegrationMessageHeaderAccessor(message).getCloseableResource(); + if (closeableResource != null) { + closeableResource.close(); + } + } + } + + }; + Iterator iterator = new Iterator() { boolean markers = FileSplitter.this.markers; diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd index 59c5d38733..a6a3adc3a8 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd @@ -833,15 +833,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 new file mode 100644 index 0000000000..2726f0d0b1 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java @@ -0,0 +1,177 @@ +/* + * 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.remote; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; + +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.rules.TemporaryFolder; + +/** + * Abstract base class for tests requiring remote file servers, e.g. (S)FTP. + * + * @author Gary Russell + * @since 4.3 + * + */ +public abstract class RemoteFileTestSupport { + + protected static int port; + + @ClassRule + public static final TemporaryFolder remoteTemporaryFolder = new TemporaryFolder(); + + @ClassRule + public static final TemporaryFolder localTemporaryFolder = new TemporaryFolder(); + + protected volatile File sourceRemoteDirectory; + + protected volatile File targetRemoteDirectory; + + protected volatile File sourceLocalDirectory; + + protected volatile File targetLocalDirectory; + + public File getSourceRemoteDirectory() { + return sourceRemoteDirectory; + } + + public File getTargetRemoteDirectory() { + return targetRemoteDirectory; + } + + public File getSourceLocalDirectory() { + return sourceLocalDirectory; + } + + public File getTargetLocalDirectory() { + return targetLocalDirectory; + } + + /** + * Default implementation creates the following folder structures: + * + * 
+	 *  $ tree remoteSource/
+	 *  remoteSource/
+	 *  ├── remoteSource1.txt - contains 'source1'
+	 *  ├── remoteSource2.txt - contains 'source2'
+	 *  ├── subRemoteSource
+	 *      ├── subRemoteSource1.txt - contains 'subSource1'
+	 *  remoteTarget/
+	 *  $ tree localSource/
+	 *  localSource/
+	 *  ├── localSource1.txt - contains 'local1'
+	 *  ├── localSource2.txt - contains 'local2'
+	 *  ├── subLocalSource
+	 *      ├── subLocalSource1.txt - contains 'subLocal1'
+	 *  localTarget/
+	 *
+ * + * The intent is tests retrieve from remoteSource and verify arrival in localTarget or send from localSource and verify + * arrival in remoteTarget. + *

+ * Subclasses can change 'remote' in these names by overriding {@link #prefix()} or override this method completely to + * create a different structure. + *

+ * While a single server exists for all tests, the directory structure is rebuilt for each test. + * @throws IOException IO Exception. + */ + @Before + public void setupFolders() throws IOException { + String prefix = prefix(); + recursiveDelete(new File(remoteTemporaryFolder.getRoot(), prefix + "Source")); + this.sourceRemoteDirectory = remoteTemporaryFolder.newFolder(prefix + "Source"); + recursiveDelete(new File(remoteTemporaryFolder.getRoot(), prefix + "Target")); + this.targetRemoteDirectory = remoteTemporaryFolder.newFolder(prefix + "Target"); + recursiveDelete(new File(localTemporaryFolder.getRoot(), "localSource")); + this.sourceLocalDirectory = localTemporaryFolder.newFolder("localSource"); + recursiveDelete(new File(localTemporaryFolder.getRoot(), "localTarget")); + this.targetLocalDirectory = localTemporaryFolder.newFolder("localTarget"); + + File file = new File(this.sourceRemoteDirectory, " " + prefix + "Source1.txt"); + file.createNewFile(); + FileOutputStream fos = new FileOutputStream(file); + fos.write("source1".getBytes()); + fos.close(); + file = new File(this.sourceRemoteDirectory, prefix + "Source2.txt"); + file.createNewFile(); + fos = new FileOutputStream(file); + fos.write("source2".getBytes()); + fos.close(); + String camelCasePrefix = camelCase(prefix); + File subSourceDirectory = new File(this.sourceRemoteDirectory, "sub" + camelCasePrefix + "Source"); + subSourceDirectory.mkdir(); + file = new File(subSourceDirectory, "sub" + camelCasePrefix + "Source1.txt"); + file.createNewFile(); + fos = new FileOutputStream(file); + fos.write("subSource1".getBytes()); + fos.close(); + file = new File(sourceLocalDirectory, "localSource1.txt"); + file.createNewFile(); + fos = new FileOutputStream(file); + fos.write("local1".getBytes()); + fos.close(); + file = new File(sourceLocalDirectory, "localSource2.txt"); + file.createNewFile(); + fos = new FileOutputStream(file); + fos.write("local2".getBytes()); + fos.close(); + File subSourceLocalDirectory = new File(this.sourceLocalDirectory, "subLocalSource"); + subSourceLocalDirectory.mkdir(); + file = new File(subSourceLocalDirectory, "subLocalSource1.txt"); + file.createNewFile(); + fos = new FileOutputStream(file); + fos.write("subLocal1".getBytes()); + fos.close(); + } + + private String camelCase(String prefix) { + char[] chars = prefix.toCharArray(); + chars[0] &= 0xdf; + return new String(chars); + } + + public void recursiveDelete(File file) { + if (file != null && file.exists()) { + File[] files = file.listFiles(); + if (files != null) { + for (File fyle : files) { + if (fyle.isDirectory()) { + recursiveDelete(fyle); + } + else { + fyle.delete(); + } + } + } + file.delete(); + } + } + + /** + * Prefix for directory/file structure; default 'remote'. + * @return the prefix. + */ + protected String prefix() { + return "remote"; + } + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java new file mode 100644 index 0000000000..e1d4aea5d7 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java @@ -0,0 +1,221 @@ +/* + * 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.remote; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Comparator; +import java.util.List; + +import org.junit.Test; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.remote.session.Session; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.integration.transformer.StreamTransformer; +import org.springframework.messaging.Message; + +/** + * @author Gary Russell + * @since 4.3 + * + */ +public class StreamingInboundTests { + + private final StreamTransformer transformer = new StreamTransformer(); + + @SuppressWarnings("unchecked") + @Test + public void testAllData() throws Exception { + Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null); + streamer.setBeanFactory(mock(BeanFactory.class)); + streamer.setRemoteDirectory("/foo"); + streamer.afterPropertiesSet(); + Message received = (Message) this.transformer.transform(streamer.receive()); + assertEquals("foo\nbar", new String(received.getPayload())); + assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); + assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + + verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource()).close(); + + received = (Message) this.transformer.transform(streamer.receive()); + assertEquals("baz\nqux", new String(received.getPayload())); + assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); + assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + + verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource()).close(); + } + + @SuppressWarnings("unchecked") + @Test + public void testLineByLine() throws Exception { + Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null); + streamer.setBeanFactory(mock(BeanFactory.class)); + streamer.setRemoteDirectory("/foo"); + streamer.afterPropertiesSet(); + QueueChannel out = new QueueChannel(); + FileSplitter splitter = new FileSplitter(); + splitter.setBeanFactory(mock(BeanFactory.class)); + splitter.setOutputChannel(out); + splitter.afterPropertiesSet(); + Message receivedStream = streamer.receive(); + splitter.handleMessage(receivedStream); + Message received = (Message) out.receive(0); + assertEquals("foo", received.getPayload()); + assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); + assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + received = (Message) out.receive(0); + assertEquals("bar", received.getPayload()); + assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); + assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + assertNull(out.receive(0)); + + verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource()).close(); + + receivedStream = streamer.receive(); + splitter.handleMessage(receivedStream); + received = (Message) out.receive(0); + assertEquals("baz", received.getPayload()); + assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); + assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + received = (Message) out.receive(0); + assertEquals("qux", received.getPayload()); + assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); + assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + assertNull(out.receive(0)); + + verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource()).close(); + } + + public static class Streamer extends AbstractRemoteFileStreamingMessageSource { + + protected Streamer(RemoteFileTemplate template, Comparator> comparator) { + super(template, comparator); + } + + @Override + public String getComponentType() { + return "Streamer"; + } + + @Override + protected List> asFileInfoList(Collection files) { + List> infos = new ArrayList>(); + for (String file : files) { + infos.add(new StringFileInfo(file)); + } + return infos; + } + + } + + public static class StringFileInfo extends AbstractFileInfo { + + private final String name; + + private StringFileInfo(String name) { + this.name = name; + } + + @Override + public boolean isDirectory() { + return false; + } + + @Override + public boolean isLink() { + return false; + } + + @Override + public long getSize() { + return 0; + } + + @Override + public long getModified() { + return 0; + } + + @Override + public String getFilename() { + return this.name.substring(this.name.lastIndexOf("/") + 1); + } + + @Override + public String getPermissions() { + return null; + } + + @Override + public String getFileInfo() { + return null; + } + + } + + public static class StringRemoteFileTemplate extends RemoteFileTemplate { + + public StringRemoteFileTemplate(SessionFactory sessionFactory) { + super(sessionFactory); + } + + } + + public static class StringSessionFactory implements SessionFactory { + + @SuppressWarnings("unchecked") + @Override + public Session getSession() { + try { + Session session = mock(Session.class); + willReturn(new String[] { "/foo/foo", "/foo/bar" }).given(session).list("/foo"); + ByteArrayInputStream foo = new ByteArrayInputStream("foo\nbar".getBytes()); + ByteArrayInputStream bar = new ByteArrayInputStream("baz\nqux".getBytes()); + willReturn(foo).given(session).readRaw("/foo/foo"); + willReturn(bar).given(session).readRaw("/foo/bar"); + + willReturn(new String[] { "/bar/foo", "/bar/bar" }).given(session).list("/bar"); + ByteArrayInputStream foo2 = new ByteArrayInputStream("foo\r\nbar".getBytes()); + ByteArrayInputStream bar2 = new ByteArrayInputStream("baz\r\nqux".getBytes()); + willReturn(foo2).given(session).readRaw("/bar/foo"); + willReturn(bar2).given(session).readRaw("/bar/bar"); + + given(session.finalizeRaw()).willReturn(true); + return session; + } + catch (Exception e) { + throw new RuntimeException("failed to mock session", e); + } + } + + } + +} 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 f306934c26..76eb9d7a71 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..06f17f2bb9 --- /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.3 + * + */ +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..f03051c1d5 --- /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.3 + * + */ +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 e3eb8e2624..0d8d6959a4 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.3.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.3.xsd index aa89f5b6a2..79571450ba 100644 --- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.3.xsd +++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-4.3.xsd @@ -24,7 +24,7 @@ - + @@ -78,38 +78,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 @@ -125,17 +95,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. - - - @@ -220,16 +163,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'. - + @@ -247,7 +203,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. - + - Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers") + 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 specify Charset (e.g., US-ASCII, ISO-8859-1, UTF-8). [UTF-8] is default + 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. @@ -559,10 +549,24 @@ + + + + + + + Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers") + + + + + + + + - + @@ -576,17 +580,6 @@ ]]> - - - - Extension used when downloading files. We - change - it right after we know it's - downloaded. - - - - @@ -597,7 +590,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..18f33270c9 --- /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.3 + */ +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/TestFtpServer.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TestFtpServer.java deleted file mode 100644 index cccd1c45ee..0000000000 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TestFtpServer.java +++ /dev/null @@ -1,260 +0,0 @@ -/* - * Copyright 2013-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; - -import java.io.File; -import java.io.FileOutputStream; -import java.io.IOException; -import java.util.Arrays; - -import javax.annotation.PostConstruct; -import javax.annotation.PreDestroy; - -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.rules.TemporaryFolder; - -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.integration.file.remote.session.CachingSessionFactory; -import org.springframework.integration.file.remote.session.SessionFactory; -import org.springframework.integration.ftp.session.DefaultFtpSessionFactory; - -/** - * Embedded FTP Server for test cases; exposes an associated session factory - * as a @Bean. - * - * @author Artem Bilan - * @author Gary Russell - * @since 3.0 - */ -@Configuration -public class TestFtpServer { - - private final TemporaryFolder ftpFolder; - - private final TemporaryFolder localFolder; - - private volatile int ftpPort; - - private volatile File ftpRootFolder; - - private volatile File sourceFtpDirectory; - - private volatile File targetFtpDirectory; - - private volatile File sourceLocalDirectory; - - private volatile File targetLocalDirectory; - - private volatile FtpServer server; - - public TestFtpServer(final String root) { - this.ftpFolder = new TemporaryFolder() { - - @Override - public void create() throws IOException { - super.create(); - ftpRootFolder = this.newFolder(root); - sourceFtpDirectory = new File(ftpRootFolder, "ftpSource"); - sourceFtpDirectory.mkdir(); - File file = new File(sourceFtpDirectory, " ftpSource1.txt"); - file.createNewFile(); - FileOutputStream fos = new FileOutputStream(file); - fos.write("source1".getBytes()); - fos.close(); - file = new File(sourceFtpDirectory, "ftpSource2.txt"); - file.createNewFile(); - fos = new FileOutputStream(file); - fos.write("source2".getBytes()); - fos.close(); - - File subSourceFtpDirectory = new File(sourceFtpDirectory, "subFtpSource"); - subSourceFtpDirectory.mkdir(); - file = new File(subSourceFtpDirectory, "subFtpSource1.txt"); - file.createNewFile(); - fos = new FileOutputStream(file); - fos.write("subSource1".getBytes()); - fos.close(); - - targetFtpDirectory = new File(ftpRootFolder, "ftpTarget"); - targetFtpDirectory.mkdir(); - } - }; - this.localFolder = new TemporaryFolder() { - - @Override - public void create() throws IOException { - super.create(); - File rootFolder = this.newFolder(root); - sourceLocalDirectory = new File(rootFolder, "localSource"); - sourceLocalDirectory.mkdirs(); - File file = new File(sourceLocalDirectory, "localSource1.txt"); - file.createNewFile(); - file = new File(sourceLocalDirectory, "localSource2.txt"); - file.createNewFile(); - - File subSourceLocalDirectory = new File(sourceLocalDirectory, "subLocalSource"); - subSourceLocalDirectory.mkdir(); - file = new File(subSourceLocalDirectory, "subLocalSource1.txt"); - file.createNewFile(); - - targetLocalDirectory = new File(rootFolder, "localTarget"); - targetLocalDirectory.mkdir(); - } - }; - } - - public File getSourceFtpDirectory() { - return sourceFtpDirectory; - } - - public File getTargetFtpDirectory() { - return targetFtpDirectory; - } - - public File getSourceLocalDirectory() { - return sourceLocalDirectory; - } - - public File getTargetLocalDirectory() { - return targetLocalDirectory; - } - - public String getTargetLocalDirectoryName() { - return targetLocalDirectory.getAbsolutePath() + File.separator; - } - - @Bean - public SessionFactory ftpSessionFactory() { - DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory(); - factory.setHost("localhost"); - factory.setPort(this.ftpPort); - factory.setUsername("foo"); - factory.setPassword("foo"); - - return new CachingSessionFactory(factory); - } - - @PostConstruct - public void before() throws Throwable { - this.ftpFolder.create(); - this.localFolder.create(); - - FtpServerFactory serverFactory = new FtpServerFactory(); - serverFactory.setUserManager(new TestUserManager(this.ftpRootFolder.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(); - this.ftpPort = listener.getPort(); - } - - - @PreDestroy - public void after() { - this.server.stop(); - this.ftpFolder.delete(); - this.localFolder.delete(); - } - - - public void recursiveDelete(File file) { - File[] files = file.listFiles(); - if (files != null) { - for (File each : files) { - recursiveDelete(each); - } - } - if (!(file.equals(this.targetFtpDirectory) || file.equals(this.targetLocalDirectory))) { - file.delete(); - } - } - - - private 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/FtpInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml index 000d22160a..46dce33a96 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml @@ -17,7 +17,6 @@ + + + + + + + + + + + + + + + + + + + + + + + + + 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/config/FtpsInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsInboundChannelAdapterParserTests-context.xml index 2722f043b9..0cac157d38 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsInboundChannelAdapterParserTests-context.xml @@ -17,9 +17,8 @@ - - - + - + 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..b3ea6ce092 --- /dev/null +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java @@ -0,0 +1,116 @@ +/* + * 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.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.session.FtpRemoteFileTemplate; +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.scheduling.support.PeriodicTrigger; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.3 + * + */ +@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 + @InboundChannelAdapter(channel = "stream") + public MessageSource ftpMessageSource() { + FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null); + messageSource.setRemoteDirectory("ftpSource/"); + 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/outbound/FtpServerOutboundTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml index 3c38a02fce..ad9f4d7180 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml @@ -9,9 +9,7 @@ http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"> - - - + @@ -23,7 +21,7 @@ request-channel="inboundGet" command="get" expression="payload" - local-directory-expression="@ftpServer.targetLocalDirectoryName + #remoteDirectory.toUpperCase()" + local-directory-expression="@extraConfig.targetLocalDirectoryName + #remoteDirectory.toUpperCase()" local-filename-generator-expression="#remoteFileName.replaceFirst('ftpSource', 'localTarget')" reply-channel="output"/> @@ -43,7 +41,7 @@ command="mget" command-options="-f" expression="payload" - local-directory-expression="@ftpServer.targetLocalDirectoryName + (#remoteDirectory ?: '')" + local-directory-expression="@extraConfig.targetLocalDirectoryName + (#remoteDirectory ?: '')" local-filename-generator-expression="#remoteFileName.replaceFirst('ftpSource', 'localTarget')" reply-channel="output"/> @@ -54,7 +52,7 @@ command="mget" expression="payload" command-options="-R" - local-directory-expression="@ftpServer.targetLocalDirectoryName + #remoteDirectory" + local-directory-expression="@extraConfig.targetLocalDirectoryName + #remoteDirectory" local-filename-generator-expression="#remoteFileName.replaceFirst('ftpSource', 'localTarget')" reply-channel="output"/> @@ -66,7 +64,7 @@ expression="payload" command-options="-R" filename-regex="(subFtpSource|.*1.txt)" - local-directory-expression="@ftpServer.targetLocalDirectoryName + #remoteDirectory" + local-directory-expression="@extraConfig.targetLocalDirectoryName + #remoteDirectory" local-filename-generator-expression="#remoteFileName.replaceFirst('ftpSource', 'localTarget')" reply-channel="output"/> diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java index f3e32a7c0d..562ba8c0d6 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java @@ -60,7 +60,9 @@ import org.mockito.stubbing.Answer; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.file.FileHeaders; @@ -71,7 +73,7 @@ import org.springframework.integration.file.remote.RemoteFileTemplate; import org.springframework.integration.file.remote.SessionCallback; 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.integration.ftp.session.FtpRemoteFileTemplate; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.PartialSuccessException; @@ -95,10 +97,7 @@ import org.springframework.util.FileCopyUtils; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD) -public class FtpServerOutboundTests { - - @Autowired - private TestFtpServer ftpServer; +public class FtpServerOutboundTests extends FtpTestSupport { @Autowired private SessionFactory ftpSessionFactory; @@ -151,10 +150,12 @@ public class FtpServerOutboundTests { @Autowired private SourcePollingChannelAdapter ftpInbound; + @Autowired + private Config config; + @Before public void setup() { - this.ftpServer.recursiveDelete(ftpServer.getTargetLocalDirectory()); - this.ftpServer.recursiveDelete(ftpServer.getTargetFtpDirectory()); + this.config.targetLocalDirectoryName = getTargetLocalDirectoryName(); } @Test @@ -325,7 +326,7 @@ public class FtpServerOutboundTests { @Test public void testInt3088MPutNotRecursive() { - this.inboundMPut.send(new GenericMessage(this.ftpServer.getSourceLocalDirectory())); + this.inboundMPut.send(new GenericMessage(getSourceLocalDirectory())); @SuppressWarnings("unchecked") Message> out = (Message>) this.output.receive(1000); assertNotNull(out); @@ -342,7 +343,7 @@ public class FtpServerOutboundTests { @Test public void testInt3088MPutRecursive() { - this.inboundMPutRecursive.send(new GenericMessage(this.ftpServer.getSourceLocalDirectory())); + this.inboundMPutRecursive.send(new GenericMessage(getSourceLocalDirectory())); @SuppressWarnings("unchecked") Message> out = (Message>) this.output.receive(1000); assertNotNull(out); @@ -365,7 +366,7 @@ public class FtpServerOutboundTests { @Test public void testInt3088MPutRecursiveFiltered() { - this.inboundMPutRecursiveFiltered.send(new GenericMessage(this.ftpServer.getSourceLocalDirectory())); + this.inboundMPutRecursiveFiltered.send(new GenericMessage(getSourceLocalDirectory())); @SuppressWarnings("unchecked") Message> out = (Message>) this.output.receive(1000); assertNotNull(out); @@ -415,7 +416,7 @@ public class FtpServerOutboundTests { assertEquals("ftpSource/", result.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); assertEquals(" ftpSource1.txt", result.getHeaders().get(FileHeaders.REMOTE_FILE)); - Session session = (Session) result.getHeaders().get(FileHeaders.REMOTE_SESSION); + Session session = (Session) result.getHeaders().get(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE); // Returned to cache assertTrue(session.isOpen()); // Raw reading is finished @@ -429,7 +430,8 @@ public class FtpServerOutboundTests { assertEquals("ftpSource/", result.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); assertEquals("ftpSource2.txt", result.getHeaders().get(FileHeaders.REMOTE_FILE)); assertSame(TestUtils.getPropertyValue(session, "targetSession"), - TestUtils.getPropertyValue(result.getHeaders().get(FileHeaders.REMOTE_SESSION), "targetSession")); + TestUtils.getPropertyValue(result.getHeaders().get(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE), + "targetSession")); } @Test @@ -504,7 +506,7 @@ public class FtpServerOutboundTests { }).when(session).write(Mockito.any(InputStream.class), Mockito.contains("localSource2")); try { - this.inboundMPut.send(new GenericMessage(this.ftpServer.getSourceLocalDirectory())); + this.inboundMPut.send(new GenericMessage(getSourceLocalDirectory())); fail("expected exception"); } catch (PartialSuccessException e) { @@ -519,7 +521,7 @@ public class FtpServerOutboundTests { @Test public void testMputRecursivePartial() throws Exception { Session session = spyOnSession(); - File sourceLocalSubDirectory = new File(ftpServer.getSourceLocalDirectory(), "subLocalSource"); + File sourceLocalSubDirectory = new File(getSourceLocalDirectory(), "subLocalSource"); assertTrue(sourceLocalSubDirectory.isDirectory()); File extra = new File(sourceLocalSubDirectory, "subLocalSource2.txt"); FileOutputStream writer = new FileOutputStream(extra); @@ -534,7 +536,7 @@ public class FtpServerOutboundTests { }).when(session).write(Mockito.any(InputStream.class), Mockito.contains("subLocalSource2")); try { - this.inboundMPutRecursive.send(new GenericMessage(this.ftpServer.getSourceLocalDirectory())); + this.inboundMPutRecursive.send(new GenericMessage(getSourceLocalDirectory())); fail("expected exception"); } catch (PartialSuccessException e) { @@ -656,6 +658,7 @@ public class FtpServerOutboundTests { } + @SuppressWarnings("unused") private static final class TestMessageSessionCallback implements MessageSessionCallback { @@ -666,4 +669,19 @@ public class FtpServerOutboundTests { } + public static class Config { + + private volatile String targetLocalDirectoryName; + + @Bean + public SessionFactory ftpSessionFactory() { + return FtpServerOutboundTests.sessionFactory(); + } + + public String getTargetLocalDirectoryName() { + return this.targetLocalDirectoryName; + } + + } + } 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 d84a53cbce..cd827bd87f 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 @@ -30,12 +30,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; @@ -43,7 +43,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; @@ -56,21 +56,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() { FtpRemoteFileTemplate template = new FtpRemoteFileTemplate(sessionFactory); @@ -143,4 +133,14 @@ public class FtpRemoteFileTemplateTests { newFile.delete(); } + @Configuration + public static class Config { + + @Bean + public SessionFactory ftpSessionFactory() { + return FtpRemoteFileTemplateTests.sessionFactory(); + } + + } + } diff --git a/spring-integration-sftp/local-test-dir/rollback/sftpSource2.txt.writing b/spring-integration-sftp/local-test-dir/rollback/sftpSource2.txt.writing new file mode 100644 index 0000000000..e69de29bb2 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..b58af344d6 --- /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.3 + * + */ +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..a75b70638b --- /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.3 + * + */ +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.3.xsd b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.3.xsd index 0a46f0fdbb..3e9813301a 100644 --- a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.3.xsd +++ b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-4.3.xsd @@ -24,7 +24,7 @@ - + Configures a 'SourcePollingChannelAdapter' Endpoint for the - 'org.springframework.integration.sftp.inbound.FtpInboundFileSynchronizingMessageSource' + '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 @@ -153,15 +99,12 @@ - + - - Allows you to provide a Regular Expression to - determine the file names - that need to be scanned. - (e.g., - "f[o]+\.txt" etc.) - + @@ -224,16 +167,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'. - + @@ -250,7 +206,7 @@ - + - + Tells this adapter if local directory must be @@ -531,32 +486,63 @@ - + - + + + + + + + + + - Identifies the directory path (e.g., - "/temp/mytransfers") - Mutually exclusive with 'remote-directory-expression'. + Identifies channel attached to this adapter. + The channel to which messages will be sent + by this adapter. - + - Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers") + 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 specify Charset (e.g., US-ASCII, - ISO-8859-1, UTF-8). [UTF-8] is default + 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. @@ -564,10 +550,24 @@ + + + + + + + Identifies the remote temporary directory path (e.g., "/remote/temp/mytransfers") + + + + + + + + - + @@ -581,16 +581,6 @@ ]]> - - - - Extension used when downloading files. We - change - it right after we know it's - downloaded. - - - @@ -601,6 +591,27 @@ + + + + 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"); + + + @@ -615,4 +626,15 @@ + + + + + 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..f14db94adc --- /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.3 + */ +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/TestSftpServer.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSftpServer.java deleted file mode 100644 index 018ccfee30..0000000000 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSftpServer.java +++ /dev/null @@ -1,186 +0,0 @@ -/* - * Copyright 2014-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.sftp; - -import java.io.File; -import java.io.FileOutputStream; -import java.io.IOException; -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.session.ServerSession; -import org.apache.sshd.server.sftp.SftpSubsystem; -import org.junit.rules.TemporaryFolder; - -import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.file.remote.session.CachingSessionFactory; -import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; - -import com.jcraft.jsch.ChannelSftp.LsEntry; - -/** - * @author Gary Russell - * @author Artem Bilan - * @since 4.1 - * - */ -public class TestSftpServer implements InitializingBean, DisposableBean { - - private final SshServer server = SshServer.setUpDefaultServer(); - - private final TemporaryFolder sftpFolder; - - private final TemporaryFolder localFolder; - - private volatile File sftpRootFolder; - - private volatile File sourceSftpDirectory; - - private volatile File targetSftpDirectory; - - private volatile File sourceLocalDirectory; - - private volatile File targetLocalDirectory; - - public TestSftpServer() { - this.sftpFolder = new TemporaryFolder() { - - @Override - public void create() throws IOException { - super.create(); - sftpRootFolder = this.newFolder("test"); - sourceSftpDirectory = new File(sftpRootFolder, "sftpSource"); - sourceSftpDirectory.mkdir(); - File file = new File(sourceSftpDirectory, "sftpSource1.txt"); - file.createNewFile(); - FileOutputStream fos = new FileOutputStream(file); - fos.write("source1".getBytes()); - fos.close(); - file = new File(sourceSftpDirectory, "sftpSource2.txt"); - file.createNewFile(); - fos = new FileOutputStream(file); - fos.write("source2".getBytes()); - fos.close(); - - File subSourceFtpDirectory = new File(sourceSftpDirectory, "subSftpSource"); - subSourceFtpDirectory.mkdir(); - file = new File(subSourceFtpDirectory, "subSftpSource1.txt"); - file.createNewFile(); - fos = new FileOutputStream(file); - fos.write("subSource1".getBytes()); - fos.close(); - - targetSftpDirectory = new File(sftpRootFolder, "sftpTarget"); - targetSftpDirectory.mkdir(); - } - }; - this.localFolder = new TemporaryFolder() { - - @Override - public void create() throws IOException { - super.create(); - File rootFolder = this.newFolder("test"); - sourceLocalDirectory = new File(rootFolder, "localSource"); - sourceLocalDirectory.mkdirs(); - File file = new File(sourceLocalDirectory, "localSource1.txt"); - file.createNewFile(); - file = new File(sourceLocalDirectory, "localSource2.txt"); - file.createNewFile(); - - File subSourceLocalDirectory = new File(sourceLocalDirectory, "subLocalSource"); - subSourceLocalDirectory.mkdir(); - file = new File(subSourceLocalDirectory, "subLocalSource1.txt"); - file.createNewFile(); - - targetLocalDirectory = new File(rootFolder, "slocalTarget"); - targetLocalDirectory.mkdir(); - } - }; - } - - @SuppressWarnings("unchecked") - @Override - public void afterPropertiesSet() throws Exception { - this.sftpFolder.create(); - this.localFolder.create(); - server.setPasswordAuthenticator(new PasswordAuthenticator() { - - @Override - public boolean authenticate(String arg0, String arg1, ServerSession arg2) { - return true; - } - - }); - server.setPort(0); - server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser")); - this.server.setSubsystemFactories(Collections.>singletonList(new SftpSubsystem.Factory())); - this.server.setFileSystemFactory(new VirtualFileSystemFactory(sftpRootFolder.getAbsolutePath())); - server.start(); - } - - @Override - public void destroy() throws Exception { - this.server.stop(true); - this.sftpFolder.delete(); - this.localFolder.delete(); - } - - public File getSourceLocalDirectory() { - return this.sourceLocalDirectory; - } - - public File getTargetLocalDirectory() { - return this.targetLocalDirectory; - } - - public String getTargetLocalDirectoryName() { - return this.targetLocalDirectory.getAbsolutePath() + File.separator; - } - - public File getTargetSftpDirectory() { - return this.targetSftpDirectory; - } - - public void recursiveDelete(File file) { - File[] files = file.listFiles(); - if (files != null) { - for (File each : files) { - recursiveDelete(each); - } - } - if (!(file.equals(this.targetSftpDirectory) || file.equals(this.targetLocalDirectory))) { - file.delete(); - } - } - - public CachingSessionFactory getSessionFactory() { - DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true); - factory.setHost("localhost"); - factory.setPort(this.server.getPort()); - factory.setUser("foo"); - factory.setPassword("foo"); - factory.setAllowUnknownKeys(true); - return new CachingSessionFactory(factory); - } - -} diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSftpServerConfig.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSftpServerConfig.java deleted file mode 100644 index 8b1db23f49..0000000000 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSftpServerConfig.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Copyright 2014-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 org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.integration.file.remote.session.CachingSessionFactory; - -import com.jcraft.jsch.ChannelSftp.LsEntry; - -/** - * @author Gary Russell - * @since 4.1 - * - */ -@Configuration -public class TestSftpServerConfig { - - @Bean - public TestSftpServer sftpServer() { - return new TestSftpServer(); - } - - @Bean - public CachingSessionFactory sftpSessionFactory(TestSftpServer server) { - return sftpServer().getSessionFactory(); - } - -} 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..f28cc9a7f9 --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java @@ -0,0 +1,117 @@ +/* + * 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.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.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.scheduling.support.PeriodicTrigger; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.jcraft.jsch.ChannelSftp.LsEntry; + +/** + * @author Gary Russell + * @since 4.3 + * + */ +@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(channel = "stream") + public MessageSource ftpMessageSource() { + SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), null); + messageSource.setRemoteDirectory("sftpSource/"); + return messageSource; + } + + @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/outbound/SftpServerOutboundTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml index cc7dcf6611..62d00206ec 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml @@ -9,6 +9,8 @@ http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"> + + @@ -19,7 +21,7 @@ request-channel="inboundGet" command="get" expression="payload" - local-directory-expression="@sftpServer.targetLocalDirectoryName + #remoteDirectory.toUpperCase()" + local-directory-expression="@extraConfig.targetLocalDirectoryName + #remoteDirectory.toUpperCase()" local-filename-generator-expression="#remoteFileName.replaceFirst('sftpSource', 'localTarget')" reply-channel="output"/> @@ -38,7 +40,7 @@ request-channel="inboundMGet" command="mget" expression="payload" - local-directory-expression="@sftpServer.targetLocalDirectoryName + #remoteDirectory" + local-directory-expression="@extraConfig.targetLocalDirectoryName + #remoteDirectory" local-filename-generator-expression="#remoteFileName.replaceFirst('sftpSource', 'localTarget')" reply-channel="output"/> @@ -49,7 +51,7 @@ command="mget" expression="payload" command-options="-R" - local-directory-expression="@sftpServer.targetLocalDirectoryName + #remoteDirectory" + local-directory-expression="@extraConfig.targetLocalDirectoryName + #remoteDirectory" local-filename-generator-expression="#remoteFileName.replaceFirst('sftpSource', 'localTarget')" reply-channel="output"/> @@ -61,7 +63,7 @@ expression="payload" command-options="-R" filename-regex="(subSftpSource|.*1.txt)" - local-directory-expression="@sftpServer.targetLocalDirectoryName + #remoteDirectory" + local-directory-expression="@extraConfig.targetLocalDirectoryName + #remoteDirectory" local-filename-generator-expression="#remoteFileName.replaceFirst('sftpSource', 'localTarget')" reply-channel="output"/> @@ -104,8 +106,6 @@ - - - * $ tree sftpSource/ - * sftpSource/ - * ??? sftpSource1.txt - contains 'source1' - * ??? sftpSource2.txt - contains 'source2' - * ??? subSftpSource - * ??? subSftpSource1.txt - contains 'subSource1' - * - * * @author Artem Bilan * @author Gary Russell * @since 3.0 */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) -public class SftpServerOutboundTests { +public class SftpServerOutboundTests extends SftpTestSupport { @Autowired private PollableChannel output; @@ -127,27 +116,25 @@ public class SftpServerOutboundTests { @Autowired private DirectChannel failing; - @Autowired - private TestSftpServer sftpServer; - @Autowired private DirectChannel inboundGetStream; @Autowired private DirectChannel inboundCallback; + @Autowired + private Config config; + @Before - @After public void setup() { - this.sftpServer.recursiveDelete(sftpServer.getTargetLocalDirectory()); - this.sftpServer.recursiveDelete(sftpServer.getTargetSftpDirectory()); + this.config.targetLocalDirectoryName = getTargetLocalDirectoryName(); } @Test public void testInt2866LocalDirectoryExpressionGET() { Session session = this.sessionFactory.getSession(); String dir = "sftpSource/"; - this.inboundGet.send(new GenericMessage(dir + "sftpSource1.txt")); + this.inboundGet.send(new GenericMessage(dir + " sftpSource1.txt")); Message result = this.output.receive(1000); assertNotNull(result); File localFile = (File) result.getPayload(); @@ -169,7 +156,7 @@ public class SftpServerOutboundTests { @Test public void testInt2866InvalidLocalDirectoryExpression() { try { - this.invalidDirExpression.send(new GenericMessage("sftpSource/sftpSource1.txt")); + this.invalidDirExpression.send(new GenericMessage("sftpSource/ sftpSource1.txt")); fail("Exception expected."); } catch (Exception e) { @@ -257,7 +244,7 @@ public class SftpServerOutboundTests { public void testInt3100RawGET() throws Exception { Session session = this.sessionFactory.getSession(); ByteArrayOutputStream baos = new ByteArrayOutputStream(); - FileCopyUtils.copy(session.readRaw("sftpSource/sftpSource1.txt"), baos); + FileCopyUtils.copy(session.readRaw("sftpSource/ sftpSource1.txt"), baos); assertTrue(session.finalizeRaw()); assertEquals("source1", new String(baos.toByteArray())); @@ -341,7 +328,7 @@ public class SftpServerOutboundTests { while (output.receive(0) != null) { // drain } - this.inboundMPut.send(new GenericMessage(this.sftpServer.getSourceLocalDirectory())); + this.inboundMPut.send(new GenericMessage(getSourceLocalDirectory())); @SuppressWarnings("unchecked") Message> out = (Message>) this.output.receive(1000); assertNotNull(out); @@ -365,7 +352,7 @@ public class SftpServerOutboundTests { while (output.receive(0) != null) { // drain } - this.inboundMPutRecursive.send(new GenericMessage(this.sftpServer.getSourceLocalDirectory())); + this.inboundMPutRecursive.send(new GenericMessage(getSourceLocalDirectory())); @SuppressWarnings("unchecked") Message> out = (Message>) this.output.receive(1000); assertNotNull(out); @@ -393,7 +380,7 @@ public class SftpServerOutboundTests { while (output.receive(0) != null) { // drain } - this.inboundMPutRecursiveFiltered.send(new GenericMessage(this.sftpServer.getSourceLocalDirectory())); + this.inboundMPutRecursiveFiltered.send(new GenericMessage(getSourceLocalDirectory())); @SuppressWarnings("unchecked") Message> out = (Message>) this.output.receive(1000); assertNotNull(out); @@ -439,12 +426,12 @@ public class SftpServerOutboundTests { session.close(); String dir = "sftpSource/"; - this.inboundGetStream.send(new GenericMessage(dir + "sftpSource1.txt")); + this.inboundGetStream.send(new GenericMessage(dir + " sftpSource1.txt")); Message result = this.output.receive(1000); assertNotNull(result); assertEquals("source1", result.getPayload()); assertEquals("sftpSource/", result.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); - assertEquals("sftpSource1.txt", result.getHeaders().get(FileHeaders.REMOTE_FILE)); + assertEquals(" sftpSource1.txt", result.getHeaders().get(FileHeaders.REMOTE_FILE)); verify(session).close(); } @@ -479,4 +466,19 @@ public class SftpServerOutboundTests { } + public static class Config { + + private volatile String targetLocalDirectoryName; + + @Bean + public SessionFactory sftpSessionFactory() { + return SftpServerOutboundTests.sessionFactory(); + } + + public String getTargetLocalDirectoryName() { + return this.targetLocalDirectoryName; + } + + } + } 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 c8f1d4933e..7a65edc141 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 @@ -22,12 +22,12 @@ 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; @@ -35,8 +35,8 @@ import org.springframework.integration.file.remote.SessionCallback; import org.springframework.integration.file.remote.SessionCallbackWithoutResult; import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; -import org.springframework.integration.sftp.TestSftpServer; -import org.springframework.integration.sftp.TestSftpServerConfig; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.sftp.SftpTestSupport; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -52,24 +52,14 @@ import com.jcraft.jsch.SftpException; * @since 4.1 * */ -@ContextConfiguration(classes = TestSftpServerConfig.class) +@ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext -public class SftpRemoteFileTemplateTests { - - @Autowired - private TestSftpServer sftpServer; +public class SftpRemoteFileTemplateTests extends SftpTestSupport { @Autowired private CachingSessionFactory 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); @@ -117,4 +107,14 @@ public class SftpRemoteFileTemplateTests { assertFalse(template.exists("foo")); } + @Configuration + public static class Config { + + @Bean + public SessionFactory ftpSessionFactory() { + return SftpRemoteFileTemplateTests.sessionFactory(); + } + + } + } diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 23d990187e..85333cee66 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -272,7 +272,8 @@ This means that (when this event is enabled), filters such as the `AcceptOnceFil meaning that, if a file with the same name appears, it will pass the filter and be sent as a message. For this purpose the `watch-events` -(`FileReadingMessageSource.setWatchEvents(FileReadingMessageSource.WatchEventType... watchEvents)`) has been introduced. +(`FileReadingMessageSource.setWatchEvents(WatchEventType... watchEvents)`) has been introduced +(`WatchEventType` is a public inner enum in `FileReadingMessageSource`). With such an option we can implement some scenarios, when we would like to do one downstream flow logic for new files, and other for modified. We can achieve that with different `` definitions, but for the same directory: @@ -858,3 +859,10 @@ public MessageHandler fileSplitter() { return splitter; } ---- + +The `FileSplitter` will also split any text-based `InputStream` into lines. +When used in conjunction with an FTP or SFTP streaming inbound channel adapter, or an FTP or SFTP outbound gateway +using the `stream` option to retrieve a file, starting with _version 4.3_, the splitter will automatically close +the session supporting the stream, when the file is completely consumed. +See <> and <> as well as <> and <> for more +information about these facilities. diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 949996f95c..7c6651bec7 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -163,7 +163,6 @@ The _FTP Inbound Channel Adapter_ is a special listener that will connect to the ---- +[[ftp-streaming]] +=== FTP Streaming Inbound Channel Adapter + +The streaming inbound channel adapter was introduced in _version 4.3_. +This adapter produces message with payloads of type `InputStream`, allowing files to be fetched without writing to the +local file system. +Since the session remains open, the consuming application is responsible for closing the session when the file has been +consumed. +The session is provided in the `closeableResource` header (`IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE`). +Standard framework components, such as the `FileSplitter` and `StreamTransformer` will automatically close the session. +See <> and <> for more information about these components. + +[source, xml] +---- + + + +---- + +Only one of `filename-pattern`, `filename-regex` or `filter` is allowed. + +IMPORTANT: Unlike the non-streaming inbound channel adapter, this adapter does not prevent duplicates by default. +If you do not delete the remote file (e.g. using an outbound gateway with an rm command) and you wish to prevent the +file being processed again, you can configure an `FtpPersistentFileListFilter` in the `filter` attribute. +If you don't actually want to persist the state, an in-memory `SimpleMetadataStore` can be used with the filter. +If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`. + [[ftp-outbound]] === FTP Outbound Channel Adapter @@ -394,7 +428,7 @@ Commands supported are: ls lists remote file(s) and supports the following options: -* -1 - just retrieve a list of filenames, default is to retrieve a list of `FileInfo` objects. +* -1 - just retrieve a list of file names, default is to retrieve a list of `FileInfo` objects. * -a - include all files (including those starting with '.') * -f - do not sort the list * -dirs - include directories (excluded by default) @@ -433,10 +467,26 @@ The remote directory is provided in the `file_remoteDirectory` header, and the f The message payload resulting from a _get_ operation is a `File` object representing the retrieved file, or an `InputStream` when the `-stream` option is provided. This option allows retrieving the file as a stream. -For text files, a common use case is to combine this operation with a <>. +For text files, a common use case is to combine this operation with a <> or +<>. When consuming remote files as streams, the user is responsible for closing the `Session` after the stream is consumed. -For convenience, the `Session` is provided in the `file_remoteSession` header. +For convenience, the `Session` is provided in the `closeableResource` header, a convenience method is provided on the +`IntegrationMessageHeaderAccessor`: + +[source, java] +---- +Closeable closeable = new IntegrationMessageHeaderAccessor(message).getCloseableResource(); +if (closeable != null) { + closeable.close(); +} +---- + +Note: In previous releases the session was in the `file_remoteSession` header, but this is deprecated - use +`closableResource` instead. + +Framework components such as the <> and <> will +automatically close the session after the data is transferred. The following shows an example of consuming a file as a stream: @@ -450,19 +500,17 @@ The following shows an example of consuming a file as a stream: remote-directory="ftpTarget" reply-channel="stream" /> - - - - - - - - + ---- -The file lines are sent to the channel `output`. +Note: if you consume the input stream in a custom component, you *must* close the `Session`. +You can either do that in your custom code, or route a copy of the message to a `service-activator` and use SpEL: + +[source, xml] +---- + +---- *mget* @@ -475,7 +523,7 @@ _mget_ retrieves multiple remote files based on a pattern and supports the follo The message payload resulting from an _mget_ operation is a `List` object - a List of File objects, each representing a retrieved file. -The remote directory is provided in the `file_remoteDirectory` header, and the pattern for the filenames is provided in the `file_remoteFile` header. +The remote directory is provided in the `file_remoteDirectory` header, and the pattern for the file names is provided in the `file_remoteFile` header. [NOTE] .Notes for when using recursion (`-R`) diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 5b06fd7966..8e56d1b70c 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -434,6 +434,41 @@ This will work for any `ResettableFileListFilter`. class="org.springframework.integration.transaction.PseudoTransactionManager" /> ---- +[[sftp-streaming]] +=== SFTP Streaming Inbound Channel Adapter + +The streaming inbound channel adapter was introduced in _version 4.3_. +This adapter produces message with payloads of type `InputStream`, allowing files to be fetched without writing to the +local file system. +Since the session remains open, the consuming application is responsible for closing the session when the file has been +consumed. +The session is provided in the `closeableResource` header (`IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE`). +Standard framework components, such as the `FileSplitter` and `StreamTransformer` will automatically close the session. +See <> and <> for more information about these components. + +[source, xml] +---- + + + +---- + +Only one of `filename-pattern`, `filename-regex` or `filter` is allowed. + +IMPORTANT: Unlike the non-streaming inbound channel adapter, this adapter does not prevent duplicates by default. +If you do not delete the remote file (e.g. using an outbound gateway with an rm command) and you wish to prevent the +file being processed again, you can configure an `SftpPersistentFileListFilter` in the `filter` attribute. +If you don't actually want to persist the state, an in-memory `SimpleMetadataStore` can be used with the filter. +If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`. + [[sftp-outbound]] === SFTP Outbound Channel Adapter @@ -539,36 +574,50 @@ The remote directory is provided in the `file_remoteDirectory` header, and the f The message payload resulting from a _get_ operation is a `File` object representing the retrieved file, or an `InputStream` when the `-stream` option is provided. This option allows retrieving the file as a stream. -For text files, a common use case is to combine this operation with a <>. +For text files, a common use case is to combine this operation with a <> or +<>. When consuming remote files as streams, the user is responsible for closing the `Session` after the stream is consumed. -For convenience, the `Session` is provided in the `file_remoteSession` header. +For convenience, the `Session` is provided in the `closeableResource` header, a convenience method is provided on the +`IntegrationMessageHeaderAccessor`: + +[source, java] +---- +Closeable closeable = new IntegrationMessageHeaderAccessor(message).getCloseableResource(); +if (closeable != null) { + closeable.close(); +} +---- + +Note: In previous releases the session was in the `file_remoteSession` header, but this is deprecated - use +`closableResource` instead. + +Framework components such as the <> and <> will +automatically close the session after the data is transferred. The following shows an example of consuming a file as a stream: [source, xml] ---- - + - - - - - - - - + ---- -The file lines are sent to the channel `output`. +Note: if you consume the input stream in a custom component, you *must* close the `Session`. +You can either do that in your custom code, or route a copy of the message to a `service-activator` and use SpEL: + +[source, xml] +---- + +---- *mget* diff --git a/src/reference/asciidoc/transformer.adoc b/src/reference/asciidoc/transformer.adoc index b121f77780..3a1c59a0e5 100644 --- a/src/reference/asciidoc/transformer.adoc +++ b/src/reference/asciidoc/transformer.adoc @@ -70,9 +70,12 @@ Just like Routers, Aggregators and other components, as of Spring Integration 2. In the above configuration we are achieving a simple transformation of the _payload_ with a simple SpEL expression and without writing a custom transformer. Our _payload_ (assuming String) will be upper-cased and concatenated with the current timestamp with some simple formatting. -_Common Transformers_ +===== Common Transformers There are also a few Transformer implementations available out of the box. + +====== Object-to-String Transformer + Because, it is fairly common to use the `toString()` representation of an Object, Spring Integration provides an `ObjectToStringTransformer` whose output is a Message with a String payload. That String is the result of invoking the toString() operation on the inbound Message's payload. [source,xml] @@ -112,7 +115,7 @@ These will use standard Java serialization by default, but you can provide an im ---- -_Object-to-Map Transformer_ +====== Object-to-Map and Map-to-Object Transformers Spring Integration also provides _Object-to-Map_ and _Map-to-Object_ transformers which utilize the Spring Expression Language (SpEL) to serialize and de-serialize the object graphs. The object hierarchy is introspected to the most primitive types (String, int, etc.). @@ -210,7 +213,35 @@ NOTE: NOTE: 'ref' and 'type' attributes are mutually exclusive. You can only use one. Also, if using the 'ref' attribute, you must point to a 'prototype' scoped bean, otherwise a BeanCreationException will be thrown.  -*JSON Transformers* +[[stream-transformer]] +====== Stream Transformer + +The `StreamTransformer` transforms `InputStream` payloads to a `byte[]` or a `String` if a `charset` is provided. + +[source, xml] +---- + + + +---- + +[source, java] +---- +@Bean +@Transformer(inputChannel = "stream", outputChannel = "data") +public StreamTransformer streamToBytes() { + return new StreamTransformer(); // transforms to byte[] +} + +@Bean +@Transformer(inputChannel = "stream", outputChannel = "data") +public StreamTransformer streamToString() { + return new StreamTransformer("UTF-8"); // transforms to String +} +---- + +====== JSON Transformers _Object to JSON_ and _JSON to Object_ transformers are provided. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e4ad90e40f..924f5e1f4e 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -27,6 +27,17 @@ The `PersistentMessageGroup`, - lazy-load proxy, - implementation is provided fo which return this instance for the `getMessageGroup()` when their `lazyLoadMessageGroups` is `true` (defaults). See <> for more information. +==== FTP/SFTP Streaming Inbound Channel Adapters + +New inbound channel adapters are provided that return an `InputStream` for each file allowing you to retrieve remote +files without writing them to the local file system +See <> and <> for more information. + +==== Stream Transformer + +A new `StreamTransformer` is provided to transform an `InputStream` payload to either a `byte[]` or `String`. +See <> for more information. + [[x4.3-general]] === General Changes @@ -122,6 +133,13 @@ See <> for more information. The outbound channel adapter can now be configured to set the destination file's lastmodified timestamp. See <> for more information. +===== Splitter Changes + +The `FileSplitter` will now automatically close an (S)FTP session when the file is completely read. +This applies when the outbound gateway returns an `InputStream` or the new (S)FTP streaming channel adapters are being +used. +See <> for more information. + ==== AMQP Changes ===== Content Type Message Converter @@ -178,6 +196,8 @@ See <> and <> for more information. ==== FTP Changes +===== Session Changes + The `FtpSession` now supports `null` for the `list()` and `listNames()` method, since it is possible by the underlying FTP Client. With that the `FtpOutboundGateway` can now be configured without `remoteDirectory` expression.