diff --git a/.gitignore b/.gitignore index 0890536..6c4962e 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,4 @@ out target spring-integration-aws/src/test/resources/awscredentials.properties *.orig +/spring-integration-java-dsl/hostkey.ser diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index 61bfb82..c29b359 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -25,13 +25,14 @@ compileTestJava { ext { activeMqVersion = '5.9.0' - embedMongoVersion = '1.45' - jacocoVersion = '0.7.0.201403182114' + apacheSshdVersion = '0.10.1' + embedMongoVersion = '1.46.0' + ftpServerVersion = '1.0.6' + jacocoVersion = '0.7.1.201405082137' jmsApiVersion = '1.1-rev-1' - log4jVersion = '1.2.17' - slf4jVersion = '1.7.6' + slf4jVersion = '1.7.7' springIntegrationVersion = '4.0.3.RELEASE' - springBootVersion = '1.1.4.RELEASE' + springBootVersion = '1.1.5.RELEASE' linkHomepage = 'https://github.com/spring-projects/spring-integration-extensions' linkCi = 'https://build.spring.io/browse/INTEXT' @@ -80,9 +81,10 @@ dependencies { testCompile "org.springframework.integration:spring-integration-test:$springIntegrationVersion" testCompile "de.flapdoodle.embed:de.flapdoodle.embed.mongo:$embedMongoVersion" + testCompile "org.apache.ftpserver:ftpserver-core:$ftpServerVersion" + testCompile "org.apache.sshd:sshd-core:$apacheSshdVersion" testCompile "org.springframework.boot:spring-boot-autoconfigure:$springBootVersion" -// testRuntime "org.slf4j:jcl-over-slf4j:$slf4jVersion" testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion" testRuntime("org.apache.activemq:activemq-broker:$activeMqVersion") testRuntime("org.apache.activemq:activemq-kahadb-store:$activeMqVersion") { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java index bb85ba8..f447036 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java @@ -31,6 +31,7 @@ import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.config.IntegrationConfigUtils; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; +import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec; import org.springframework.integration.dsl.support.MessageChannelReference; @@ -132,12 +133,25 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean } else if (component instanceof SourcePollingChannelAdapterSpec) { SourcePollingChannelAdapterSpec spec = (SourcePollingChannelAdapterSpec) component; - String id = ((IntegrationComponentSpec) spec).getId(); SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1(); + String id = ((IntegrationComponentSpec) spec).getId(); if (!StringUtils.hasText(id)) { id = generateBeanName(pollingChannelAdapterFactoryBean); } registerComponent(pollingChannelAdapterFactoryBean, id); + + MessageSource messageSource = spec.get().getT2(); + if (!this.beanFactory + .getBeansOfType(MessageSource.class, false, false) + .values() + .contains(messageSource)) { + String messageSourceId = id + ".source"; + if (messageSource instanceof NamedComponent + && ((NamedComponent) messageSource).getComponentName() != null) { + messageSourceId = ((NamedComponent) messageSource).getComponentName(); + } + registerComponent(messageSource, messageSourceId); + } } else if (!this.beanFactory .getBeansOfType(AopUtils.getTargetClass(component), false, false) diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileTransferringMessageHandlerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileTransferringMessageHandlerSpec.java new file mode 100644 index 0000000..ff74fee --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileTransferringMessageHandlerSpec.java @@ -0,0 +1,152 @@ +/* + * Copyright 2014 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.dsl.file; + +import java.lang.reflect.Constructor; +import java.nio.charset.Charset; +import java.util.Collection; +import java.util.Collections; + +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.dsl.core.ComponentsRegistration; +import org.springframework.integration.dsl.core.MessageHandlerSpec; +import org.springframework.integration.file.DefaultFileNameGenerator; +import org.springframework.integration.file.FileNameGenerator; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.file.remote.handler.FileTransferringMessageHandler; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +/** + * @author Artem Bilan + */ +public abstract class FileTransferringMessageHandlerSpec> + extends MessageHandlerSpec> + implements ComponentsRegistration { + + private FileNameGenerator fileNameGenerator; + + private DefaultFileNameGenerator defaultFileNameGenerator; + + protected FileTransferringMessageHandlerSpec(SessionFactory sessionFactory) { + this.target = new FileTransferringMessageHandler(sessionFactory); + } + + protected FileTransferringMessageHandlerSpec(RemoteFileTemplate remoteFileTemplate) { + this.target = new FileTransferringMessageHandler(remoteFileTemplate); + } + + @SuppressWarnings("unchecked") + protected FileTransferringMessageHandlerSpec(RemoteFileTemplate remoteFileTemplate, FileExistsMode fileExistsMode) { + Constructor fileExistsModeConstructor = + ClassUtils.getConstructorIfAvailable(FileTransferringMessageHandler.class, RemoteFileTemplate.class, + FileExistsMode.class); + if (fileExistsModeConstructor == null) { + logger.warn("The 'FileExistsMode' constructor argument for the 'FileTransferringMessageHandler' is " + + "available since Spring Integration 4.1. Will be ignored for previous versions."); + this.target = new FileTransferringMessageHandler(remoteFileTemplate); + } + else { + try { + this.target = (FileTransferringMessageHandler) fileExistsModeConstructor.newInstance(remoteFileTemplate, + fileExistsMode); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + } + + public S autoCreateDirectory(boolean autoCreateDirectory) { + this.target.setAutoCreateDirectory(autoCreateDirectory); + return _this(); + } + + public S remoteFileSeparator(String remoteFileSeparator) { + this.target.setRemoteFileSeparator(remoteFileSeparator); + return _this(); + } + + public S remoteDirectory(String remoteDirectory) { + this.target.setRemoteDirectoryExpression(new LiteralExpression(remoteDirectory)); + return _this(); + } + + public S remoteDirectoryExpression(String remoteDirectoryExpression) { + this.target.setRemoteDirectoryExpression(PARSER.parseExpression(remoteDirectoryExpression)); + return _this(); + } + + public S temporaryRemoteDirectory(String temporaryRemoteDirectory) { + this.target.setTemporaryRemoteDirectoryExpression(new LiteralExpression(temporaryRemoteDirectory)); + return _this(); + } + + public S temporaryRemoteDirectoryExpression(String temporaryRemoteDirectoryExpression) { + this.target.setTemporaryRemoteDirectoryExpression(PARSER.parseExpression(temporaryRemoteDirectoryExpression)); + return _this(); + } + + public S useTemporaryFileName(boolean useTemporaryFileName) { + this.target.setUseTemporaryFileName(useTemporaryFileName); + return _this(); + } + + public S fileNameGenerator(FileNameGenerator fileNameGenerator) { + this.target.setFileNameGenerator(fileNameGenerator); + return _this(); + } + + public S fileNameGeneratorExpression(String fileNameGeneratorExpression) { + Assert.isNull(this.fileNameGenerator, + "'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive."); + this.defaultFileNameGenerator = new DefaultFileNameGenerator(); + this.defaultFileNameGenerator.setExpression(fileNameGeneratorExpression); + return fileNameGenerator(this.defaultFileNameGenerator); + } + + public S charset(String charset) { + this.target.setCharset(charset); + return _this(); + } + + public S charset(Charset charset) { + this.target.setCharset(charset.name()); + return _this(); + } + + public S temporaryFileSuffix(String temporaryFileSuffix) { + this.target.setTemporaryFileSuffix(temporaryFileSuffix); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + if (this.defaultFileNameGenerator != null) { + return Collections.singletonList(this.defaultFileNameGenerator); + } + return Collections.emptyList(); + } + + @Override + protected FileTransferringMessageHandler doGet() { + throw new UnsupportedOperationException(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java index 9995ea6..6c97aa8 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java @@ -77,7 +77,7 @@ public class FileWritingMessageHandlerSpec "'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive."); this.defaultFileNameGenerator = new DefaultFileNameGenerator(); this.defaultFileNameGenerator.setExpression(fileNameGeneratorExpression); - return _this(); + return fileNameGenerator(this.defaultFileNameGenerator); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileOutboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileOutboundGatewaySpec.java new file mode 100644 index 0000000..4851215 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileOutboundGatewaySpec.java @@ -0,0 +1,127 @@ +/* + * Copyright 2014 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.dsl.file; + +import java.io.File; + +import org.springframework.integration.dsl.core.MessageHandlerSpec; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.filters.RegexPatternFileListFilter; +import org.springframework.integration.file.filters.SimplePatternFileListFilter; +import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + */ +public abstract class RemoteFileOutboundGatewaySpec> + extends MessageHandlerSpec> { + + private FileListFilter filter; + + private FileListFilter mputFilter; + + protected RemoteFileOutboundGatewaySpec(AbstractRemoteFileOutboundGateway outboundGateway) { + this.target = outboundGateway; + } + + public S options(String options) { + this.target.setOptions(options); + return _this(); + } + + public S options(AbstractRemoteFileOutboundGateway.Option... options) { + Assert.noNullElements(options); + StringBuilder optionsString = new StringBuilder(); + for (AbstractRemoteFileOutboundGateway.Option option : options) { + optionsString.append(option.getOption()).append(" "); + } + this.target.setOptions(optionsString.toString()); + return _this(); + } + + public S remoteFileSeparator(String remoteFileSeparator) { + this.target.setRemoteFileSeparator(remoteFileSeparator); + return _this(); + } + + public S localDirectory(java.io.File localDirectory) { + this.target.setLocalDirectory(localDirectory); + return _this(); + } + + public S localDirectoryExpression(String localDirectoryExpression) { + this.target.setLocalDirectoryExpression(PARSER.parseExpression(localDirectoryExpression)); + return _this(); + } + + public S autoCreateLocalDirectory(boolean autoCreateLocalDirectory) { + this.target.setAutoCreateLocalDirectory(autoCreateLocalDirectory); + return _this(); + } + + public S temporaryFileSuffix(String temporaryFileSuffix) { + this.target.setTemporaryFileSuffix(temporaryFileSuffix); + return _this(); + } + + public S filter(FileListFilter filter) { + Assert.isNull(this.filter, + "The 'filter' (" + this.filter + ") is already configured for the: " + this); + this.filter = filter; + this.target.setFilter(filter); + return _this(); + } + + public abstract S patternFileNameFilter(String pattern); + + public abstract S regexFileNameFilter(String regex); + + public S mputFilter(FileListFilter filter) { + Assert.isNull(this.mputFilter, + "The 'filter' (" + this.mputFilter + ") is already configured for the: " + this); + this.mputFilter = filter; + this.target.setMputFilter(filter); + return _this(); + } + + + public S patternMputFilter(String pattern) { + return mputFilter(new SimplePatternFileListFilter(pattern)); + } + + public S regexMpuFilter(String regex) { + return mputFilter(new RegexPatternFileListFilter(regex)); + } + + + public S renameExpression(String expression) { + this.target.setRenameExpression(expression); + return _this(); + } + + public S localFilenameGeneratorExpression(String localFilenameGeneratorExpression) { + this.target.setLocalFilenameGeneratorExpression(PARSER.parseExpression(localFilenameGeneratorExpression)); + return _this(); + } + + @Override + protected AbstractRemoteFileOutboundGateway doGet() { + throw new UnsupportedOperationException(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteInboundChannelAdapterSpec.java new file mode 100644 index 0000000..be84053 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteInboundChannelAdapterSpec.java @@ -0,0 +1,112 @@ +/* + * Copyright 2014 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.dsl.file; + +import java.io.File; +import java.util.Collection; +import java.util.Collections; + +import org.springframework.integration.dsl.core.ComponentsRegistration; +import org.springframework.integration.dsl.core.MessageSourceSpec; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer; +import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + */ +public abstract class RemoteInboundChannelAdapterSpec, + MS extends AbstractInboundFileSynchronizingMessageSource> + extends MessageSourceSpec implements ComponentsRegistration { + + protected final AbstractInboundFileSynchronizer synchronizer; + + private FileListFilter filter; + + protected RemoteInboundChannelAdapterSpec(AbstractInboundFileSynchronizer synchronizer) { + this.synchronizer = synchronizer; + } + + public S autoCreateLocalDirectory(boolean autoCreateLocalDirectory) { + this.target.setAutoCreateLocalDirectory(autoCreateLocalDirectory); + return _this(); + } + + public S localDirectory(File localDirectory) { + this.target.setLocalDirectory(localDirectory); + return _this(); + } + + public S localFilter(FileListFilter localFileListFilter) { + this.target.setLocalFilter(localFileListFilter); + return _this(); + } + + public S remoteFileSeparator(String remoteFileSeparator) { + this.synchronizer.setRemoteFileSeparator(remoteFileSeparator); + return _this(); + } + + public S localFilenameGeneratorExpression(String localFilenameGeneratorExpression) { + this.synchronizer.setLocalFilenameGeneratorExpression(PARSER.parseExpression(localFilenameGeneratorExpression)); + return _this(); + } + + public S temporaryFileSuffix(String temporaryFileSuffix) { + this.synchronizer.setTemporaryFileSuffix(temporaryFileSuffix); + return _this(); + } + + public S remoteDirectory(String remoteDirectory) { + this.synchronizer.setRemoteDirectory(remoteDirectory); + return _this(); + } + + public S filter(FileListFilter filter) { + Assert.isNull(this.filter, + "The 'filter' (" + this.filter + ") is already configured for the: " + this); + this.filter = filter; + this.synchronizer.setFilter(filter); + return _this(); + } + + public abstract S patternFilter(String pattern); + + public abstract S regexFilter(String regex); + + public S deleteRemoteFiles(boolean deleteRemoteFiles) { + this.synchronizer.setDeleteRemoteFiles(deleteRemoteFiles); + return _this(); + } + + public S preserveTimestamp(boolean preserveTimestamp) { + this.synchronizer.setPreserveTimestamp(preserveTimestamp); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + return Collections.singletonList(this.synchronizer); + } + + @Override + protected MS doGet() { + throw new UnsupportedOperationException(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/Ftp.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/Ftp.java new file mode 100644 index 0000000..9ec3648 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/Ftp.java @@ -0,0 +1,71 @@ +/* + * Copyright 2014 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.dsl.ftp; + +import java.util.Comparator; + +import org.apache.commons.net.ftp.FTPFile; + +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.integration.ftp.gateway.FtpOutboundGateway; + +/** + * @author Artem Bilan + */ +public abstract class Ftp { + + public static FtpInboundChannelAdapterSpec inboundAdapter(SessionFactory sessionFactory) { + return inboundAdapter(sessionFactory, null); + } + + public static FtpInboundChannelAdapterSpec inboundAdapter(SessionFactory sessionFactory, + Comparator receptionOrderComparator) { + return new FtpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator); + } + + public static FtpMessageHandlerSpec outboundAdapter(SessionFactory sessionFactory) { + return new FtpMessageHandlerSpec(sessionFactory); + } + + public static FtpMessageHandlerSpec outboundAdapter(SessionFactory sessionFactory, + FileExistsMode fileExistsMode) { + return outboundAdapter(new RemoteFileTemplate(sessionFactory), fileExistsMode); + } + + public static FtpMessageHandlerSpec outboundAdapter(RemoteFileTemplate remoteFileTemplate) { + return new FtpMessageHandlerSpec(remoteFileTemplate); + } + + public static FtpMessageHandlerSpec outboundAdapter(RemoteFileTemplate remoteFileTemplate, + FileExistsMode fileExistsMode) { + return new FtpMessageHandlerSpec(remoteFileTemplate, fileExistsMode); + } + + public static FtpOutboundGatewaySpec outboundGateway(SessionFactory sessionFactory, + AbstractRemoteFileOutboundGateway.Command command, String expression) { + return outboundGateway(sessionFactory, command.getCommand(), expression); + } + + public static FtpOutboundGatewaySpec outboundGateway(SessionFactory sessionFactory, + String command, String expression) { + return new FtpOutboundGatewaySpec(new FtpOutboundGateway(sessionFactory, command, expression)); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpInboundChannelAdapterSpec.java new file mode 100644 index 0000000..046ba8d --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpInboundChannelAdapterSpec.java @@ -0,0 +1,53 @@ +/* + * Copyright 2014 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.dsl.ftp; + +import java.io.File; +import java.util.Comparator; + +import org.apache.commons.net.ftp.FTPFile; + +import org.springframework.integration.dsl.file.RemoteInboundChannelAdapterSpec; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; +import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; +import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer; +import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource; + +/** + * @author Artem Bilan + */ +public class FtpInboundChannelAdapterSpec + extends RemoteInboundChannelAdapterSpec { + + FtpInboundChannelAdapterSpec(SessionFactory sessionFactory, Comparator comparator) { + super(new FtpInboundFileSynchronizer(sessionFactory)); + this.target = new FtpInboundFileSynchronizingMessageSource(this.synchronizer, comparator); + } + + @Override + public FtpInboundChannelAdapterSpec patternFilter(String pattern) { + return filter(new FtpSimplePatternFileListFilter(pattern)); + } + + @Override + public FtpInboundChannelAdapterSpec regexFilter(String regex) { + return filter(new FtpRegexPatternFileListFilter(regex)); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpMessageHandlerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpMessageHandlerSpec.java new file mode 100644 index 0000000..7e301f1 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpMessageHandlerSpec.java @@ -0,0 +1,43 @@ +/* + * Copyright 2014 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.dsl.ftp; + +import org.apache.commons.net.ftp.FTPFile; + +import org.springframework.integration.dsl.file.FileTransferringMessageHandlerSpec; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.support.FileExistsMode; + +/** + * @author Artem Bilan + */ +public class FtpMessageHandlerSpec extends FileTransferringMessageHandlerSpec { + + FtpMessageHandlerSpec(SessionFactory sessionFactory) { + super(sessionFactory); + } + + FtpMessageHandlerSpec(RemoteFileTemplate remoteFileTemplate) { + super(remoteFileTemplate); + } + + FtpMessageHandlerSpec(RemoteFileTemplate remoteFileTemplate, FileExistsMode fileExistsMode) { + super(remoteFileTemplate, fileExistsMode); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpOutboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpOutboundGatewaySpec.java new file mode 100644 index 0000000..578dca8 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpOutboundGatewaySpec.java @@ -0,0 +1,45 @@ +/* + * Copyright 2014 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.dsl.ftp; + +import org.apache.commons.net.ftp.FTPFile; + +import org.springframework.integration.dsl.file.RemoteFileOutboundGatewaySpec; +import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; +import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; +import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; + +/** + * @author Artem Bilan + */ +public class FtpOutboundGatewaySpec extends RemoteFileOutboundGatewaySpec { + + FtpOutboundGatewaySpec(AbstractRemoteFileOutboundGateway outboundGateway) { + super(outboundGateway); + } + + @Override + public FtpOutboundGatewaySpec patternFileNameFilter(String pattern) { + return filter(new FtpSimplePatternFileListFilter(pattern)); + } + + @Override + public FtpOutboundGatewaySpec regexFileNameFilter(String regex) { + return filter(new FtpRegexPatternFileListFilter(regex)); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/package-info.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/package-info.java new file mode 100644 index 0000000..8eba86f --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides FTP Components support for Spring Integration Java DSL. + */ +package org.springframework.integration.dsl.ftp; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/Sftp.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/Sftp.java new file mode 100644 index 0000000..2f862cc --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/Sftp.java @@ -0,0 +1,71 @@ +/* + * Copyright 2014 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.dsl.sftp; + +import java.util.Comparator; + +import com.jcraft.jsch.ChannelSftp; + +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.integration.sftp.gateway.SftpOutboundGateway; + +/** + * @author Artem Bilan + */ +public abstract class Sftp { + + public static SftpInboundChannelAdapterSpec inboundAdapter(SessionFactory sessionFactory) { + return inboundAdapter(sessionFactory, null); + } + + public static SftpInboundChannelAdapterSpec inboundAdapter(SessionFactory sessionFactory, + Comparator receptionOrderComparator) { + return new SftpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator); + } + + public static SftpMessageHandlerSpec outboundAdapter(SessionFactory sessionFactory) { + return new SftpMessageHandlerSpec(sessionFactory); + } + + public static SftpMessageHandlerSpec outboundAdapter(SessionFactory sessionFactory, + FileExistsMode fileExistsMode) { + return outboundAdapter(new RemoteFileTemplate(sessionFactory), fileExistsMode); + } + + public static SftpMessageHandlerSpec outboundAdapter(RemoteFileTemplate remoteFileTemplate) { + return new SftpMessageHandlerSpec(remoteFileTemplate); + } + + public static SftpMessageHandlerSpec outboundAdapter(RemoteFileTemplate remoteFileTemplate, + FileExistsMode fileExistsMode) { + return new SftpMessageHandlerSpec(remoteFileTemplate, fileExistsMode); + } + + public static SftpOutboundGatewaySpec outboundGateway(SessionFactory sessionFactory, + AbstractRemoteFileOutboundGateway.Command command, String expression) { + return outboundGateway(sessionFactory, command.getCommand(), expression); + } + + public static SftpOutboundGatewaySpec outboundGateway(SessionFactory sessionFactory, + String command, String expression) { + return new SftpOutboundGatewaySpec(new SftpOutboundGateway(sessionFactory, command, expression)); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpInboundChannelAdapterSpec.java new file mode 100644 index 0000000..3295bfe --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpInboundChannelAdapterSpec.java @@ -0,0 +1,53 @@ +/* + * Copyright 2014 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.dsl.sftp; + +import java.io.File; +import java.util.Comparator; + +import com.jcraft.jsch.ChannelSftp; + +import org.springframework.integration.dsl.file.RemoteInboundChannelAdapterSpec; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter; +import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter; +import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer; +import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource; + +/** + * @author Artem Bilan + */ +public class SftpInboundChannelAdapterSpec + extends RemoteInboundChannelAdapterSpec { + + SftpInboundChannelAdapterSpec(SessionFactory sessionFactory, Comparator comparator) { + super(new SftpInboundFileSynchronizer(sessionFactory)); + this.target = new SftpInboundFileSynchronizingMessageSource(this.synchronizer, comparator); + } + + @Override + public SftpInboundChannelAdapterSpec patternFilter(String pattern) { + return filter(new SftpSimplePatternFileListFilter(pattern)); + } + + @Override + public SftpInboundChannelAdapterSpec regexFilter(String regex) { + return filter(new SftpRegexPatternFileListFilter(regex)); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpMessageHandlerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpMessageHandlerSpec.java new file mode 100644 index 0000000..5e86d51 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpMessageHandlerSpec.java @@ -0,0 +1,43 @@ +/* + * Copyright 2014 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.dsl.sftp; + +import com.jcraft.jsch.ChannelSftp; + +import org.springframework.integration.dsl.file.FileTransferringMessageHandlerSpec; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.support.FileExistsMode; + +/** + * @author Artem Bilan + */ +public class SftpMessageHandlerSpec extends FileTransferringMessageHandlerSpec { + + SftpMessageHandlerSpec(SessionFactory sessionFactory) { + super(sessionFactory); + } + + SftpMessageHandlerSpec(RemoteFileTemplate remoteFileTemplate) { + super(remoteFileTemplate); + } + + SftpMessageHandlerSpec(RemoteFileTemplate remoteFileTemplate, FileExistsMode fileExistsMode) { + super(remoteFileTemplate, fileExistsMode); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpOutboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpOutboundGatewaySpec.java new file mode 100644 index 0000000..e0ed014 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpOutboundGatewaySpec.java @@ -0,0 +1,45 @@ +/* + * Copyright 2014 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.dsl.sftp; + +import com.jcraft.jsch.ChannelSftp; + +import org.springframework.integration.dsl.file.RemoteFileOutboundGatewaySpec; +import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; +import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter; +import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter; + +/** + * @author Artem Bilan + */ +public class SftpOutboundGatewaySpec extends RemoteFileOutboundGatewaySpec { + + + SftpOutboundGatewaySpec(AbstractRemoteFileOutboundGateway outboundGateway) { + super(outboundGateway); + } + + @Override + public SftpOutboundGatewaySpec patternFileNameFilter(String pattern) { + return filter(new SftpSimplePatternFileListFilter(pattern)); + } + + @Override + public SftpOutboundGatewaySpec regexFileNameFilter(String regex) { + return filter(new SftpRegexPatternFileListFilter(regex)); + } +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/package-info.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/package-info.java new file mode 100644 index 0000000..b8cff3d --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides SFTP Components support for Spring Integration Java DSL. + */ +package org.springframework.integration.dsl.sftp; diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index ba7d9ef..390239d 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -16,7 +16,10 @@ package org.springframework.integration.dsl.test; -import static org.hamcrest.Matchers.*; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.endsWith; +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.isOneOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -27,7 +30,6 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import java.io.FileInputStream; -import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; import java.nio.charset.Charset; @@ -42,12 +44,24 @@ import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.regex.Matcher; +import com.jcraft.jsch.ChannelSftp; +import com.mongodb.MongoClient; +import de.flapdoodle.embed.mongo.MongodExecutable; +import de.flapdoodle.embed.mongo.MongodStarter; +import de.flapdoodle.embed.mongo.config.MongodConfigBuilder; +import de.flapdoodle.embed.mongo.config.Net; +import de.flapdoodle.embed.mongo.distribution.Version; +import de.flapdoodle.embed.process.runtime.Network; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; +import org.apache.commons.net.ftp.FTPFile; import org.hamcrest.Matchers; +import org.junit.After; import org.junit.AfterClass; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; import org.junit.runner.RunWith; @@ -95,7 +109,9 @@ import org.springframework.integration.dsl.amqp.Amqp; import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; import org.springframework.integration.dsl.file.File; +import org.springframework.integration.dsl.ftp.Ftp; import org.springframework.integration.dsl.jms.Jms; +import org.springframework.integration.dsl.sftp.Sftp; import org.springframework.integration.dsl.support.Pollers; import org.springframework.integration.dsl.support.Transformers; import org.springframework.integration.endpoint.MethodInvokingMessageSource; @@ -103,11 +119,15 @@ import org.springframework.integration.event.core.MessagingEvent; import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; +import org.springframework.integration.ftp.session.DefaultFtpSessionFactory; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice; import org.springframework.integration.mongodb.store.MongoDbChannelMessageStore; import org.springframework.integration.router.MethodInvokingRouter; import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.PriorityCapableChannelMessageStore; import org.springframework.integration.store.SimpleMessageStore; @@ -140,14 +160,6 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.StreamUtils; -import com.mongodb.MongoClient; -import de.flapdoodle.embed.mongo.MongodExecutable; -import de.flapdoodle.embed.mongo.MongodStarter; -import de.flapdoodle.embed.mongo.config.MongodConfigBuilder; -import de.flapdoodle.embed.mongo.config.Net; -import de.flapdoodle.embed.mongo.distribution.Version; -import de.flapdoodle.embed.process.runtime.Network; - /** * @author Artem Bilan * @author Tim Ysewyn @@ -1102,6 +1114,160 @@ public class IntegrationFlowTests { } + @Autowired + private TestFtpServer ftpServer; + + @Autowired + private DefaultFtpSessionFactory ftpSessionFactory; + + @Autowired + private TestSftpServer sftpServer; + + @Autowired + private DefaultSftpSessionFactory sftpSessionFactory; + + @Before + @After + public void setupRemoteFileServers() { + this.ftpServer.recursiveDelete(this.ftpServer.getTargetLocalDirectory()); + this.ftpServer.recursiveDelete(this.ftpServer.getTargetFtpDirectory()); + this.sftpServer.recursiveDelete(this.sftpServer.getTargetLocalDirectory()); + this.sftpServer.recursiveDelete(this.sftpServer.getTargetSftpDirectory()); + } + + @Autowired + @Qualifier("ftpInboundResultChannel") + private PollableChannel ftpInboundResultChannel; + + @Test + public void testFtpInboundFlow() { + Message message = this.ftpInboundResultChannel.receive(1000); + assertNotNull(message); + Object payload = message.getPayload(); + assertThat(payload, instanceOf(java.io.File.class)); + java.io.File file = (java.io.File) payload; + assertThat(file.getName(), isOneOf("FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a")); + assertThat(file.getAbsolutePath(), containsString("ftpTest")); + + message = this.ftpInboundResultChannel.receive(1000); + assertNotNull(message); + file = (java.io.File) message.getPayload(); + assertThat(file.getName(), isOneOf("FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a")); + assertThat(file.getAbsolutePath(), containsString("ftpTest")); + + this.controlBus.send("@ftpInboundAdapter.stop()"); + } + + @Autowired + @Qualifier("sftpInboundResultChannel") + private PollableChannel sftpInboundResultChannel; + + @Test + public void testSftpInboundFlow() { + Message message = this.sftpInboundResultChannel.receive(1000); + assertNotNull(message); + Object payload = message.getPayload(); + assertThat(payload, instanceOf(java.io.File.class)); + java.io.File file = (java.io.File) payload; + assertThat(file.getName(), isOneOf("SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a")); + assertThat(file.getAbsolutePath(), containsString("sftpTest")); + + message = this.sftpInboundResultChannel.receive(1000); + assertNotNull(message); + file = (java.io.File) message.getPayload(); + assertThat(file.getName(), isOneOf("SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a")); + assertThat(file.getAbsolutePath(), containsString("sftpTest")); + + this.controlBus.send("@sftpInboundAdapter.stop()"); + } + + @Autowired + @Qualifier("toFtpChannel") + private MessageChannel toFtpChannel; + + @Test + public void testFtpOutboundFlow() { + String fileName = "foo.file"; + this.toFtpChannel.send(MessageBuilder.withPayload("foo") + .setHeader(FileHeaders.FILENAME, fileName) + .build()); + + RemoteFileTemplate template = new RemoteFileTemplate<>(this.ftpSessionFactory); + FTPFile[] files = template.execute(session -> + session.list(this.ftpServer.getTargetFtpDirectory().getName() + "/" + fileName)); + assertEquals(1, files.length); + assertEquals(3, files[0].getSize()); + } + + @Autowired + @Qualifier("toSftpChannel") + private MessageChannel toSftpChannel; + + @Test + public void testSftpOutboundFlow() { + String fileName = "foo.file"; + this.toSftpChannel.send(MessageBuilder.withPayload("foo") + .setHeader(FileHeaders.FILENAME, fileName) + .build()); + + RemoteFileTemplate template = new RemoteFileTemplate<>(this.sftpSessionFactory); + ChannelSftp.LsEntry[] files = template.execute(session -> + session.list(this.sftpServer.getTargetSftpDirectory().getName() + "/" + fileName)); + assertEquals(1, files.length); + assertEquals(3, files[0].getAttrs().getSize()); + } + + @Autowired + @Qualifier("ftpMgetInputChannel") + private MessageChannel ftpMgetInputChannel; + + @Autowired + @Qualifier("remoteFileOutputChannel") + private PollableChannel remoteFileOutputChannel; + + @Test + @SuppressWarnings("unchecked") + public void testFtpMgetFlow() { + String dir = "ftpSource/"; + this.ftpMgetInputChannel.send(new GenericMessage(dir + "*")); + Message result = this.remoteFileOutputChannel.receive(1000); + assertNotNull(result); + List localFiles = (List) result.getPayload(); + // should have filtered ftpSource2.txt + assertEquals(2, localFiles.size()); + + for (java.io.File file : localFiles) { + assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"), + Matchers.containsString(dir)); + } + assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"), + Matchers.containsString(dir + "subFtpSource")); + } + + + @Autowired + @Qualifier("sftpMgetInputChannel") + private MessageChannel sftpMgetInputChannel; + + @Test + @SuppressWarnings("unchecked") + public void testSftpMgetFlow() { + String dir = "sftpSource/"; + this.sftpMgetInputChannel.send(new GenericMessage(dir + "*")); + Message result = this.remoteFileOutputChannel.receive(1000); + assertNotNull(result); + List localFiles = (List) result.getPayload(); + // should have filtered sftpSource2.txt + assertEquals(2, localFiles.size()); + + for (java.io.File file : localFiles) { + assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"), + Matchers.containsString(dir)); + } + assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(java.io.File.separator), "/"), + Matchers.containsString(dir + "subSftpSource")); + } + @MessagingGateway(defaultRequestChannel = "controlBus") private static interface ControlBusGateway { @@ -1144,7 +1310,7 @@ public class IntegrationFlowTests { @Bean(name = PollerMetadata.DEFAULT_POLLER) public PollerMetadata poller() { - return Pollers.fixedRate(500).get(); + return Pollers.fixedRate(500).maxMessagesPerPoll(1).get(); } @Bean(name = IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME) @@ -1214,6 +1380,96 @@ public class IntegrationFlowTests { .transform(String::toUpperCase) .get(); } + + @Autowired + private TestFtpServer ftpServer; + + @Autowired + private DefaultFtpSessionFactory ftpSessionFactory; + + @Autowired + private TestSftpServer sftpServer; + + @Autowired + private DefaultSftpSessionFactory sftpSessionFactory; + + @Bean + public IntegrationFlow ftpInboundFlow() { + return IntegrationFlows + .from(Ftp.inboundAdapter(this.ftpSessionFactory) + .preserveTimestamp(true) + .remoteDirectory("ftpSource") + .regexFilter(".*\\.txt$") + .localFilenameGeneratorExpression("#this.toUpperCase() + '.a'") + .localDirectory(this.ftpServer.getTargetLocalDirectory()), + e -> e.id("ftpInboundAdapter")) + .channel(MessageChannels.queue("ftpInboundResultChannel")) + .get(); + } + + @Bean + public IntegrationFlow sftpInboundFlow() { + return IntegrationFlows + .from(Sftp.inboundAdapter(this.sftpSessionFactory) + .preserveTimestamp(true) + .remoteDirectory("sftpSource") + .regexFilter(".*\\.txt$") + .localFilenameGeneratorExpression("#this.toUpperCase() + '.a'") + .localDirectory(this.sftpServer.getTargetLocalDirectory()), + e -> e.id("sftpInboundAdapter")) + .channel(MessageChannels.queue("sftpInboundResultChannel")) + .get(); + } + + @Bean + public IntegrationFlow ftpOutboundFlow() { + return IntegrationFlows.from("toFtpChannel") + .handle(Ftp.outboundAdapter(this.ftpSessionFactory) + .useTemporaryFileName(false) + .remoteDirectory(this.ftpServer.getTargetFtpDirectory().getName()) + ).get(); + } + + @Bean + public IntegrationFlow sftpOutboundFlow() { + return IntegrationFlows.from("toSftpChannel") + .handle(Sftp.outboundAdapter(this.sftpSessionFactory) + .useTemporaryFileName(false) + .remoteDirectory(this.sftpServer.getTargetSftpDirectory().getName()) + ).get(); + } + + @Bean + public PollableChannel remoteFileOutputChannel() { + return new QueueChannel(); + } + + @Bean + public IntegrationFlow ftpMGetFlow() { + return IntegrationFlows.from("ftpMgetInputChannel") + .handle(Ftp.outboundGateway(this.ftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET, + "payload") + .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE) + .regexFileNameFilter("(subFtpSource|.*1.txt)") + .localDirectoryExpression("@ftpServer.targetLocalDirectoryName + #remoteDirectory") + .localFilenameGeneratorExpression("#remoteFileName.replaceFirst('ftpSource', 'localTarget')")) + .channel(remoteFileOutputChannel()) + .get(); + } + + @Bean + public IntegrationFlow sftpMGetFlow() { + return IntegrationFlows.from("sftpMgetInputChannel") + .handle(Sftp.outboundGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET, + "payload") + .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE) + .regexFileNameFilter("(subSftpSource|.*1.txt)") + .localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory") + .localFilenameGeneratorExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')")) + .channel(remoteFileOutputChannel()) + .get(); + } + } @Configuration diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestFtpServer.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestFtpServer.java new file mode 100644 index 0000000..fcac1dc --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestFtpServer.java @@ -0,0 +1,253 @@ +/* + * Copyright 2014 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.dsl.test; + +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.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.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.ftp.session.DefaultFtpSessionFactory; +import org.springframework.integration.test.util.SocketUtils; + +/** + * Embedded FTP Server for test cases; exposes an associated session factory + * as a @Bean. + * + * @author Artem Bilan + * @author Gary Russell + */ +@Configuration("ftpServer") +public class TestFtpServer { + + private final int ftpPort = SocketUtils.findAvailableServerSocket(); + + private final TemporaryFolder ftpFolder; + + private final TemporaryFolder localFolder; + + 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() { + this.ftpFolder = new TemporaryFolder() { + + @Override + public void create() throws IOException { + super.create(); + ftpRootFolder = this.newFolder("ftpTest"); + 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("ftpTest"); + 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 DefaultFtpSessionFactory ftpSessionFactory() { + DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory(); + factory.setHost("localhost"); + factory.setPort(this.ftpPort); + factory.setUsername("foo"); + factory.setPassword("foo"); + return 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(ftpPort); + serverFactory.addListener("default", factory.createListener()); + + server = serverFactory.createServer(); + server.start(); + } + + + @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-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestSftpServer.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestSftpServer.java new file mode 100644 index 0000000..850bc9c --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestSftpServer.java @@ -0,0 +1,195 @@ +/* + * Copyright 2014 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.dsl.test; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.Arrays; + +import org.apache.sshd.SshServer; +import org.apache.sshd.common.NamedFactory; +import org.apache.sshd.common.file.FileSystemView; +import org.apache.sshd.common.file.nativefs.NativeFileSystemFactory; +import org.apache.sshd.common.file.nativefs.NativeFileSystemView; +import org.apache.sshd.server.Command; +import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider; +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.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; +import org.springframework.util.SocketUtils; + +/** + * @author Gary Russell + * @author Artem Bilan + */ +@Configuration("sftpServer") +public class TestSftpServer implements InitializingBean, DisposableBean { + + private final SshServer server = SshServer.setUpDefaultServer(); + + private final int port = SocketUtils.findAvailableTcpPort(); + + 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("sftpTest"); + 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("sftpTest"); + 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(); + } + }; + } + + @SuppressWarnings("unchecked") + @Override + public void afterPropertiesSet() throws Exception { + this.sftpFolder.create(); + this.localFolder.create(); + this.server.setPasswordAuthenticator((username, password, session) -> true); + this.server.setPort(this.port); + this.server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser")); + SftpSubsystem.Factory sftp = new SftpSubsystem.Factory(); + this.server.setSubsystemFactories(Arrays.>asList(sftp)); + this.server.setFileSystemFactory(new NativeFileSystemFactory() { + + @Override + public FileSystemView createFileSystemView(org.apache.sshd.common.Session session) { + return new NativeFileSystemView(session.getUsername(), false) { + + @Override + public String getVirtualUserDir() { + return sftpRootFolder.getAbsolutePath(); + } + }; + } + + }); + this.server.start(); + } + + @Override + public void destroy() throws Exception { + this.server.stop(); + 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(); + } + } + + @Bean + public DefaultSftpSessionFactory sftpSessionFactory() { + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true); + factory.setHost("localhost"); + factory.setPort(this.port); + factory.setUser("foo"); + factory.setPassword("foo"); + return factory; + } + +}