From be9767b8a9c7ac4c26e0f9c1c1cdb63b7665a361 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 29 Oct 2014 23:50:42 +0200 Subject: [PATCH] DSL: Introduce `SubFlow`s Add `subFlow` support for `router`s Move some tests to separate domain classes Fix some component registrations DSL: Fix `SubFlow` registration according PR discussion * Add IO plugin * Fix SI 4.0 <-> 4.1 compatibility by `springIoCheck` results * Further divide for tests to separate domain classes DSL: Add `discardFlow` to the `.filter()` Make `.router()` `subFlow`s as `always return to the main flow` Add `.publishSubscribeChannel()` EIP-method with `.subscriber()` to specify subscribers as `subFlow`s Polishing --- spring-integration-java-dsl/build.gradle | 19 +- .../integration/dsl/Channels.java | 27 +- .../integration/dsl/FilterEndpointSpec.java | 28 +- .../dsl/IntegrationFlowBuilder.java | 46 +- .../dsl/IntegrationFlowDefinition.java | 109 ++- .../integration/dsl/PublishSubscribeSpec.java | 60 ++ .../integration/dsl/RouterSpec.java | 80 +- .../dsl/StandardIntegrationFlow.java | 41 + .../dsl/channel/MessageChannels.java | 25 +- .../channel/PublishSubscribeChannelSpec.java | 28 +- .../dsl/channel/QueueChannelSpec.java | 27 +- .../IntegrationFlowBeanPostProcessor.java | 16 +- .../integration/dsl/test/amqp/AmqpTests.java | 147 +++ .../integration/dsl/test/file/FileTests.java | 261 +++++ .../dsl/test/flows/IntegrationFlowTests.java | 904 ++---------------- .../integration/dsl/test/ftp/FtpTests.java | 259 +++++ .../dsl/test/{ => ftp}/TestFtpServer.java | 2 +- .../integration/dsl/test/jdbc/JdbcTests.java | 8 +- .../integration/dsl/test/jms/JmsTests.java | 263 +++++ .../dsl/test/mongodb/MongoDbTest.java | 199 ++++ .../integration/dsl/test/sftp/SftpTests.java | 250 +++++ .../dsl/test/{ => sftp}/TestSftpServer.java | 2 +- 22 files changed, 1841 insertions(+), 960 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java create mode 100644 spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java create mode 100644 spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/file/FileTests.java create mode 100644 spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/ftp/FtpTests.java rename spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/{ => ftp}/TestFtpServer.java (99%) create mode 100644 spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jms/JmsTests.java create mode 100644 spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mongodb/MongoDbTest.java create mode 100644 spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/sftp/SftpTests.java rename spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/{ => sftp}/TestSftpServer.java (99%) diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index e020027..82bc98a 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -5,15 +5,32 @@ apply from: "${rootProject.projectDir}/publish-maven.gradle" apply plugin: 'eclipse' apply plugin: 'idea' +buildscript { + repositories { + maven { url 'http://repo.spring.io/plugins-release' } + } + dependencies { + classpath 'org.springframework.build.gradle:spring-io-plugin:0.0.3.RELEASE' + } +} + group = 'org.springframework.integration' repositories { - if (version.endsWith('BUILD-SNAPSHOT')) { + if (version.endsWith('BUILD-SNAPSHOT') || project.hasProperty('platformVersion')) { maven { url 'http://repo.spring.io/libs-snapshot' } } maven { url 'http://repo.spring.io/libs-milestone' } } +if (project.hasProperty('platformVersion')) { + apply plugin: 'spring-io' + + dependencies { + springIoVersions "io.spring.platform:platform-versions:${platformVersion}@properties" + } +} + compileJava { sourceCompatibility = 1.6 targetCompatibility = 1.6 diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Channels.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Channels.java index 0abd121..a5c1f25 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Channels.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Channels.java @@ -16,7 +16,7 @@ package org.springframework.integration.dsl; -import java.util.concurrent.BlockingQueue; +import java.util.Queue; import java.util.concurrent.Executor; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -68,11 +68,11 @@ public class Channels { return MessageChannels.queue(id, capacity); } - public QueueChannelSpec queue(BlockingQueue> queue) { + public QueueChannelSpec queue(Queue> queue) { return MessageChannels.queue(queue); } - public QueueChannelSpec queue(String id, BlockingQueue> queue) { + public QueueChannelSpec queue(String id, Queue> queue) { return MessageChannels.queue(id, queue); } @@ -110,14 +110,23 @@ public class Channels { return MessageChannels.rendezvous(id); } - public PublishSubscribeChannelSpec publishSubscribe() { + public PublishSubscribeChannelSpec> publishSubscribe() { return MessageChannels.publishSubscribe(); } - public PublishSubscribeChannelSpec publishSubscribe(Executor executor) { + public PublishSubscribeChannelSpec> publishSubscribe(Executor executor) { return MessageChannels.publishSubscribe(executor); } + public PublishSubscribeChannelSpec> publishSubscribe(String id, + Executor executor) { + return MessageChannels.publishSubscribe(id, executor); + } + + public PublishSubscribeChannelSpec> publishSubscribe(String id) { + return MessageChannels.publishSubscribe(id); + } + public ExecutorChannelSpec executor(Executor executor) { return MessageChannels.executor(executor); } @@ -126,14 +135,6 @@ public class Channels { return MessageChannels.executor(id, executor); } - public PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) { - return MessageChannels.publishSubscribe(id, executor); - } - - public PublishSubscribeChannelSpec publishSubscribe(String id) { - return MessageChannels.publishSubscribe(id); - } - public AmqpPollableMessageChannelSpec> amqpPollable( ConnectionFactory connectionFactory) { return Amqp.pollableChannel(connectionFactory); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java index 7a2a172..7ed1d1f 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java @@ -16,14 +16,23 @@ package org.springframework.integration.dsl; +import java.util.Collection; +import java.util.Collections; + +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.dsl.core.ComponentsRegistration; import org.springframework.integration.dsl.core.ConsumerEndpointSpec; import org.springframework.integration.filter.MessageFilter; import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; /** * @author Artem Bilan */ -public final class FilterEndpointSpec extends ConsumerEndpointSpec { +public final class FilterEndpointSpec extends ConsumerEndpointSpec + implements ComponentsRegistration { + + private IntegrationFlow discardFlow; FilterEndpointSpec(MessageFilter messageFilter) { super(messageFilter); @@ -44,9 +53,26 @@ public final class FilterEndpointSpec extends ConsumerEndpointSpec getComponentsToRegister() { + if (this.discardFlow != null) { + return Collections.singletonList(this.discardFlow); + } + return null; + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index abacec6..1383336 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -16,56 +16,14 @@ package org.springframework.integration.dsl; -import java.util.Set; - -import org.springframework.beans.factory.BeanCreationException; -import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype; - /** * @author Artem Bilan */ public final class IntegrationFlowBuilder extends IntegrationFlowDefinition { + @Override public StandardIntegrationFlow get() { - if (this.currentMessageChannel instanceof FixedSubscriberChannelPrototype) { - throw new BeanCreationException("The 'currentMessageChannel' (" + this.currentMessageChannel + - ") is a prototype for FixedSubscriberChannel which can't be created without MessageHandler " + - "constructor argument. That means that '.fixedSubscriberChannel()' can't be the last EIP-method " + - "in the IntegrationFlow definition."); - } - - if (this.integrationComponents.size() == 1) { - if (this.currentComponent != null) { - if (this.currentComponent instanceof SourcePollingChannelAdapterSpec) { - throw new BeanCreationException("The 'SourcePollingChannelAdapter' (" + this.currentComponent - + ") " + "must be configured with at least one 'MessageChanel' or 'MessageHandler'."); - } - } - else if (this.currentMessageChannel != null) { - throw new BeanCreationException("The 'IntegrationFlow' can't consist of only one 'MessageChannel'. " + - "Add at lest '.bridge()' EIP-method before the end of flow."); - } - } - return new StandardIntegrationFlow(this.integrationComponents); - } - - public static final class StandardIntegrationFlow implements IntegrationFlow { - - private final Set integrationComponents; - - StandardIntegrationFlow(Set integrationComponents) { - this.integrationComponents = integrationComponents; - } - - public Set getIntegrationComponents() { - return integrationComponents; - } - - @Override - public void accept(IntegrationFlowDefinition flow) { - throw new UnsupportedOperationException(); - } - + return super.get(); } } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index 2d1a8a7..e8c756d 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -20,6 +20,7 @@ import java.util.Collection; import java.util.LinkedHashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.Executor; import org.springframework.aop.framework.Advised; import org.springframework.aop.support.AopUtils; @@ -35,6 +36,7 @@ import org.springframework.integration.config.SourcePollingChannelAdapterFactory import org.springframework.integration.core.GenericSelector; import org.springframework.integration.core.MessageSelector; import org.springframework.integration.dsl.channel.MessageChannelSpec; +import org.springframework.integration.dsl.core.ComponentsRegistration; import org.springframework.integration.dsl.core.ConsumerEndpointSpec; import org.springframework.integration.dsl.core.MessageHandlerSpec; import org.springframework.integration.dsl.support.BeanNameMessageProcessor; @@ -77,6 +79,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; /** @@ -116,24 +119,15 @@ public abstract class IntegrationFlowDefinition(new BridgeHandler()), null); - } - this.currentMessageChannel = messageChannel; - return this.registerOutputChannelIfCan(this.currentMessageChannel); + return channel(new MessageChannelReference(messageChannelName)); } public B channel(Function> channels) { @@ -143,7 +137,28 @@ public abstract class IntegrationFlowDefinition messageChannelSpec) { Assert.notNull(messageChannelSpec); - return this.channel(messageChannelSpec.get()); + return channel(messageChannelSpec.get()); + } + + public B channel(MessageChannel messageChannel) { + Assert.notNull(messageChannel); + if (this.currentMessageChannel != null) { + this.register(new GenericEndpointSpec(new BridgeHandler()), null); + } + this.currentMessageChannel = messageChannel; + return registerOutputChannelIfCan(this.currentMessageChannel); + } + + public B publishSubscribeChannel(Consumer publishSubscribeChannelConfigurer) { + return publishSubscribeChannel(null, publishSubscribeChannelConfigurer); + } + + public B publishSubscribeChannel(Executor executor, + Consumer publishSubscribeChannelConfigurer) { + Assert.notNull(publishSubscribeChannelConfigurer); + PublishSubscribeSpec spec = new PublishSubscribeSpec(executor); + publishSubscribeChannelConfigurer.accept(spec); + return addComponents(spec.getComponentsToRegister()).channel(spec); } public B controlBus() { @@ -269,6 +284,9 @@ public abstract class IntegrationFlowDefinition B handle(MessageHandlerSpec messageHandlerSpec, Consumer> endpointConfigurer) { Assert.notNull(messageHandlerSpec); + if (messageHandlerSpec instanceof ComponentsRegistration) { + addComponents(((ComponentsRegistration) messageHandlerSpec).getComponentsToRegister()); + } return handle(messageHandlerSpec.get(), endpointConfigurer); } @@ -422,7 +440,7 @@ public abstract class IntegrationFlowDefinition B route(R router, Consumer> routerConfigurer, Consumer> endpointConfigurer) { + Collection componentsToRegister = null; if (routerConfigurer != null) { RouterSpec routerSpec = new RouterSpec(router); routerConfigurer.accept(routerSpec); + componentsToRegister = routerSpec.getComponentsToRegister(); } - return this.route(router, endpointConfigurer); + + route(router, endpointConfigurer); + + final MessageChannel afterRouterChannel = new DirectChannel(); + boolean hasSubFlows = false; + if (!CollectionUtils.isEmpty(componentsToRegister)) { + for (Object component : componentsToRegister) { + if (component instanceof IntegrationFlowDefinition) { + hasSubFlows = true; + IntegrationFlowDefinition flowBuilder = (IntegrationFlowDefinition) component; + addComponent(flowBuilder.fixedSubscriberChannel() + .bridge(new Consumer>() { + + @Override + public void accept(GenericEndpointSpec bridge) { + bridge.get().getT2().setOutputChannel(afterRouterChannel); + } + + }) + .get()); + } + else { + addComponent(component); + } + } + } + if (hasSubFlows) { + channel(afterRouterChannel); + } + return _this(); } public B routeToRecipients(Consumer routerConfigurer) { @@ -626,6 +675,9 @@ public abstract class IntegrationFlowDefinition + implements ComponentsRegistration { + + private final List subscriberFlows = new ArrayList(); + + PublishSubscribeSpec() { + super(); + } + + PublishSubscribeSpec(Executor executor) { + super(executor); + } + + @Override + public PublishSubscribeSpec id(String id) { + return super.id(id); + } + + public PublishSubscribeSpec subscribe(IntegrationFlow flow) { + IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(this.channel); + flow.accept(flowBuilder); + this.subscriberFlows.add(flowBuilder.get()); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + return this.subscriberFlows; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/RouterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/RouterSpec.java index bdd38e9..1a610a1 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/RouterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/RouterSpec.java @@ -16,12 +16,35 @@ package org.springframework.integration.dsl; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import javax.annotation.PostConstruct; + +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.dsl.core.ComponentsRegistration; import org.springframework.integration.router.AbstractMappingMessageRouter; +import org.springframework.integration.router.MappingMessageRouterManagement; +import org.springframework.integration.support.context.NamedComponent; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * @author Artem Bilan */ -public final class RouterSpec extends AbstractRouterSpec, R> { +public final class RouterSpec extends AbstractRouterSpec, R> + implements ComponentsRegistration { + + private final List subFlows = new ArrayList(); + + private String prefix; + + private String suffix; + + private RouterSubFlowMappingProvider mappingProvider; RouterSpec(R router) { super(router); @@ -33,18 +56,73 @@ public final class RouterSpec extends Ab } public RouterSpec prefix(String prefix) { + Assert.state(this.subFlows.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); + this.prefix = prefix; this.target.setPrefix(prefix); return _this(); } public RouterSpec suffix(String suffix) { + Assert.state(this.subFlows.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); + this.suffix = suffix; this.target.setSuffix(suffix); return _this(); } public RouterSpec channelMapping(String key, String channelName) { + Assert.hasText(key); + Assert.hasText(channelName); this.target.setChannelMapping(key, channelName); return _this(); } + public RouterSpec subFlowMapping(String key, IntegrationFlow subFlow) { + Assert.hasText(key); + Assert.notNull(subFlow); + Assert.state(!(StringUtils.hasText(this.prefix) || StringUtils.hasText(this.suffix)), + "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); + + DirectChannel channel = new DirectChannel(); + IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel); + subFlow.accept(flowBuilder); + + this.subFlows.add(flowBuilder); + + if (this.mappingProvider == null) { + this.mappingProvider = new RouterSubFlowMappingProvider(this.target); + this.subFlows.add(this.mappingProvider); + } + this.mappingProvider.addMapping(key, channel); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + return this.subFlows; + } + + private static class RouterSubFlowMappingProvider { + + private final MappingMessageRouterManagement router; + + private final Map mapping = new HashMap(); + + public RouterSubFlowMappingProvider(MappingMessageRouterManagement router) { + this.router = router; + } + + void addMapping(String key, NamedComponent channel) { + this.mapping.put(key, channel); + } + + @PostConstruct + public void init() { + for (Map.Entry entry : this.mapping.entrySet()) { + this.router.setChannelMapping(entry.getKey(), entry.getValue().getComponentName()); + + } + } + + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java new file mode 100644 index 0000000..3ff1dce --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java @@ -0,0 +1,41 @@ +/* + * 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; + +import java.util.Set; + +/** +* @author Artem Bilan +*/ +public class StandardIntegrationFlow implements IntegrationFlow { + + private final Set integrationComponents; + + StandardIntegrationFlow(Set integrationComponents) { + this.integrationComponents = integrationComponents; + } + + public Set getIntegrationComponents() { + return integrationComponents; + } + + @Override + public void accept(IntegrationFlowDefinition flow) { + throw new UnsupportedOperationException(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java index acb63d9..aa9679a 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java @@ -16,7 +16,7 @@ package org.springframework.integration.dsl.channel; -import java.util.concurrent.BlockingQueue; +import java.util.Queue; import java.util.concurrent.Executor; import org.springframework.integration.store.ChannelMessageStore; @@ -44,11 +44,11 @@ public final class MessageChannels { return queue().id(id); } - public static QueueChannelSpec queue(BlockingQueue> queue) { + public static QueueChannelSpec queue(Queue> queue) { return new QueueChannelSpec(queue); } - public static QueueChannelSpec queue(String id, BlockingQueue> queue) { + public static QueueChannelSpec queue(String id, Queue> queue) { return queue(queue).id(id); } @@ -103,20 +103,23 @@ public final class MessageChannels { return queue(messageGroupStore, groupId).id(id); } - public static PublishSubscribeChannelSpec publishSubscribe() { - return new PublishSubscribeChannelSpec(); + public static > PublishSubscribeChannelSpec publishSubscribe() { + return new PublishSubscribeChannelSpec(); } - public static PublishSubscribeChannelSpec publishSubscribe(String id) { - return publishSubscribe().id(id); + public static > PublishSubscribeChannelSpec publishSubscribe( + String id) { + return MessageChannels.publishSubscribe().id(id); } - public static PublishSubscribeChannelSpec publishSubscribe(Executor executor) { - return new PublishSubscribeChannelSpec(executor); + public static > PublishSubscribeChannelSpec publishSubscribe( + Executor executor) { + return new PublishSubscribeChannelSpec(executor); } - public static PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) { - return publishSubscribe(executor).id(id); + public static > PublishSubscribeChannelSpec publishSubscribe(String id, + Executor executor) { + return MessageChannels.publishSubscribe(executor).id(id); } private MessageChannels() { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.java index d93dad6..2cdbae9 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.java @@ -24,40 +24,40 @@ import org.springframework.util.ErrorHandler; /** * @author Artem Bilan */ -public class PublishSubscribeChannelSpec - extends MessageChannelSpec { +public class PublishSubscribeChannelSpec> + extends MessageChannelSpec { - PublishSubscribeChannelSpec() { + protected PublishSubscribeChannelSpec() { this.channel = new PublishSubscribeChannel(); } - PublishSubscribeChannelSpec(Executor executor) { + protected PublishSubscribeChannelSpec(Executor executor) { this.channel = new PublishSubscribeChannel(executor); } - PublishSubscribeChannelSpec errorHandler(ErrorHandler errorHandler) { + public S errorHandler(ErrorHandler errorHandler) { this.channel.setErrorHandler(errorHandler); - return this; + return _this(); } - public PublishSubscribeChannelSpec ignoreFailures(boolean ignoreFailures) { + public S ignoreFailures(boolean ignoreFailures) { this.channel.setIgnoreFailures(ignoreFailures); - return this; + return _this(); } - public PublishSubscribeChannelSpec applySequence(boolean applySequence) { + public S applySequence(boolean applySequence) { this.channel.setApplySequence(applySequence); - return this; + return _this(); } - public PublishSubscribeChannelSpec maxSubscribers(Integer maxSubscribers) { + public S maxSubscribers(Integer maxSubscribers) { this.channel.setMaxSubscribers(maxSubscribers); - return this; + return _this(); } - public PublishSubscribeChannelSpec minSubscribers(int minSubscribers) { + public S minSubscribers(int minSubscribers) { this.channel.setMinSubscribers(minSubscribers); - return this; + return _this(); } } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java index 220e9b7..6c58d09 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java @@ -16,6 +16,8 @@ package org.springframework.integration.dsl.channel; +import java.lang.reflect.Constructor; +import java.util.Queue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.locks.Lock; @@ -24,20 +26,21 @@ import org.springframework.integration.store.ChannelMessageStore; import org.springframework.integration.store.MessageGroupQueue; import org.springframework.integration.store.PriorityCapableChannelMessageStore; import org.springframework.messaging.Message; +import org.springframework.util.ClassUtils; /** * @author Artem Bilan */ public class QueueChannelSpec extends MessageChannelSpec { - protected BlockingQueue> queue; + protected Queue> queue; protected Integer capacity; QueueChannelSpec() { } - QueueChannelSpec(BlockingQueue> queue) { + QueueChannelSpec(Queue> queue) { this.queue = queue; } @@ -48,7 +51,25 @@ public class QueueChannelSpec extends MessageChannelSpec queueConstructor = + ClassUtils.getConstructorIfAvailable(QueueChannel.class, Queue.class); + if (queueConstructor == null) { + if (!(this.queue instanceof BlockingQueue)) { + throw new IllegalArgumentException("The 'queue' must be an instance of BlockingQueue " + + "for Spring Integration versions less than 4.1"); + } + else { + this.channel = new QueueChannel((BlockingQueue>) this.queue); + } + } + else { + try { + this.channel = (QueueChannel) queueConstructor.newInstance(this.queue); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } } else if (this.capacity != null) { this.channel = new QueueChannel(this.capacity); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/IntegrationFlowBeanPostProcessor.java index 23d75e5..999b0f9 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/IntegrationFlowBeanPostProcessor.java @@ -36,8 +36,8 @@ import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlowBuilder; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec; +import org.springframework.integration.dsl.StandardIntegrationFlow; import org.springframework.integration.dsl.core.ConsumerEndpointSpec; -import org.springframework.integration.dsl.core.IntegrationComponentSpec; import org.springframework.integration.dsl.support.MessageChannelReference; import org.springframework.integration.support.context.NamedComponent; import org.springframework.messaging.MessageChannel; @@ -69,8 +69,8 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean @Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { - if (bean instanceof IntegrationFlowBuilder.StandardIntegrationFlow) { - return processStandardIntegrationFlow((IntegrationFlowBuilder.StandardIntegrationFlow) bean, beanName); + if (bean instanceof StandardIntegrationFlow) { + return processStandardIntegrationFlow((StandardIntegrationFlow) bean, beanName); } else if (bean instanceof IntegrationFlow) { return processIntegrationFlowImpl((IntegrationFlow) bean, beanName); @@ -78,9 +78,10 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean return bean; } - private Object processStandardIntegrationFlow(IntegrationFlowBuilder.StandardIntegrationFlow flow, + private Object processStandardIntegrationFlow(StandardIntegrationFlow flow, String beanName) { String flowNamePrefix = beanName + "."; + int subFlowNameIndex = 0; int channelNameIndex = 0; for (Object component : flow.getIntegrationComponents()) { if (component instanceof ConsumerEndpointSpec) { @@ -147,7 +148,7 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean else if (component instanceof SourcePollingChannelAdapterSpec) { SourcePollingChannelAdapterSpec spec = (SourcePollingChannelAdapterSpec) component; SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1(); - String id = ((IntegrationComponentSpec) spec).getId(); + String id = spec.getId(); if (!StringUtils.hasText(id)) { id = generateBeanName(pollingChannelAdapterFactoryBean); } @@ -166,6 +167,11 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean registerComponent(messageSource, messageSourceId); } } + else if (component instanceof StandardIntegrationFlow) { + String subFlowBeanName = flowNamePrefix + "subFlow" + + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + subFlowNameIndex++; + registerComponent(component, subFlowBeanName); + } else if (!this.beanFactory .getBeansOfType(AopUtils.getTargetClass(component), false, false) .values() diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java new file mode 100644 index 0000000..678bce2 --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java @@ -0,0 +1,147 @@ +/* + * 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.amqp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.AnonymousQueue; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.MessageProducers; +import org.springframework.integration.dsl.amqp.Amqp; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class AmqpTests { + + @Autowired + private AmqpTemplate amqpTemplate; + + @Autowired + @Qualifier("queue") + private Queue amqpQueue; + + @Test + public void testAmqpInboundGatewayFlow() throws Exception { + Object result = this.amqpTemplate.convertSendAndReceive(this.amqpQueue.getName(), "world"); + assertEquals("HELLO WORLD", result); + } + + @Autowired + @Qualifier("amqpOutboundInput") + private MessageChannel amqpOutboundInput; + + @Autowired + @Qualifier("amqpReplyChannel.channel") + private PollableChannel amqpReplyChannel; + + @Test + public void testAmqpOutboundFlow() throws Exception { + this.amqpOutboundInput.send(MessageBuilder.withPayload("hello through the amqp") + .setHeader("routingKey", "foo") + .build()); + Message receive = null; + int i = 0; + do { + receive = this.amqpReplyChannel.receive(); + if (receive != null) { + break; + } + Thread.sleep(100); + i++; + } while (i < 10); + + assertNotNull(receive); + assertEquals("HELLO THROUGH THE AMQP", receive.getPayload()); + } + + @Configuration + @EnableAutoConfiguration + public static class ContextConfiguration { + + @Autowired + private ConnectionFactory rabbitConnectionFactory; + + @Autowired + private AmqpTemplate amqpTemplate; + + @Bean + public Queue queue() { + return new AnonymousQueue(); + } + + @Bean + public IntegrationFlow amqpFlow() { + return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue())) + .transform("hello "::concat) + .transform(String.class, String::toUpperCase) + .get(); + } + + @Bean + public IntegrationFlow amqpOutboundFlow() { + return IntegrationFlows.from(Amqp.channel("amqpOutboundInput", this.rabbitConnectionFactory)) + .handle(Amqp.outboundAdapter(this.amqpTemplate).routingKeyExpression("headers.routingKey")) + .get(); + } + + @Bean + public Queue fooQueue() { + return new Queue("foo"); + } + + @Bean + public Queue amqpReplyChannel() { + return new Queue("amqpReplyChannel"); + } + + @Bean + public IntegrationFlow amqpInboundFlow() { + return IntegrationFlows.from((MessageProducers p) -> p.amqp(this.rabbitConnectionFactory, fooQueue())) + .transform(String.class, String::toUpperCase) + .channel(Amqp.pollableChannel(this.rabbitConnectionFactory) + .queueName("amqpReplyChannel") + .channelTransacted(true)) + .get(); + } + + } + +} diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/file/FileTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/file/FileTests.java new file mode 100644 index 0000000..c48d6b3 --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/file/FileTests.java @@ -0,0 +1,261 @@ +/* + * 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.file; + +import static org.hamcrest.Matchers.endsWith; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.List; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.aop.TargetSource; +import org.springframework.aop.framework.Advised; +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.MessageProducers; +import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.core.Pollers; +import org.springframework.integration.dsl.support.Transformers; +import org.springframework.integration.file.DefaultFileNameGenerator; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.FileWritingMessageHandler; +import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHandlingException; +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; +import org.springframework.util.StreamUtils; + +/** + * @author Artem Bilan + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class FileTests { + + private static final File tmpDir = new File(System.getProperty("java.io.tmpdir")); + + @Autowired + private ListableBeanFactory beanFactory; + + @Autowired + private ControlBusGateway controlBus; + + @Autowired + @Qualifier("fileFlow1Input") + private MessageChannel fileFlow1Input; + + @Autowired + @Qualifier("fileWriting.handler") + private MessageHandler fileWritingMessageHandler; + + @Autowired + @Qualifier("tailChannel") + private PollableChannel tailChannel; + + @Autowired + private ApacheCommonsFileTailingMessageProducer tailer; + + @Autowired + @Qualifier("fileReadingResultChannel") + private PollableChannel fileReadingResultChannel; + + @Autowired + @Qualifier("fileWritingInput") + private MessageChannel fileWritingInput; + + @Autowired + @Qualifier("fileWritingResultChannel") + private PollableChannel fileWritingResultChannel; + + @Test + public void testFileHandler() throws Exception { + Message message = MessageBuilder.withPayload("foo").setHeader(FileHeaders.FILENAME, "foo").build(); + try { + this.fileFlow1Input.send(message); + fail("NullPointerException expected"); + } + catch (Exception e) { + assertThat(e, instanceOf(MessageHandlingException.class)); + assertThat(e.getCause(), instanceOf(NullPointerException.class)); + } + DefaultFileNameGenerator fileNameGenerator = new DefaultFileNameGenerator(); + fileNameGenerator.setBeanFactory(this.beanFactory); + Object targetFileWritingMessageHandler = this.fileWritingMessageHandler; + if (this.fileWritingMessageHandler instanceof Advised) { + TargetSource targetSource = ((Advised) this.fileWritingMessageHandler).getTargetSource(); + if (targetSource != null) { + targetFileWritingMessageHandler = targetSource.getTarget(); + } + } + DirectFieldAccessor dfa = new DirectFieldAccessor(targetFileWritingMessageHandler); + dfa.setPropertyValue("fileNameGenerator", fileNameGenerator); + this.fileFlow1Input.send(message); + + assertTrue(new File(tmpDir, "foo").exists()); + } + + @Test + public void testMessageProducerFlow() throws Exception { + FileOutputStream file = new FileOutputStream(new File(tmpDir, "TailTest")); + for (int i = 0; i < 50; i++) { + file.write((i + "\n").getBytes()); + } + this.tailer.start(); + for (int i = 0; i < 50; i++) { + Message message = this.tailChannel.receive(5000); + assertNotNull(message); + assertEquals("hello " + i, message.getPayload()); + } + assertNull(this.tailChannel.receive(1)); + + this.controlBus.send("@tailer.stop()"); + file.close(); + } + + + @Test + public void testFileReadingFlow() throws Exception { + List evens = new ArrayList<>(25); + for (int i = 0; i < 50; i++) { + boolean even = i % 2 == 0; + String extension = even ? ".sitest" : ".foofile"; + if (even) { + evens.add(i); + } + FileOutputStream file = new FileOutputStream(new File(tmpDir, i + extension)); + file.write(("" + i).getBytes()); + file.flush(); + file.close(); + } + + Message message = fileReadingResultChannel.receive(10000); + assertNotNull(message); + Object payload = message.getPayload(); + assertThat(payload, instanceOf(List.class)); + @SuppressWarnings("unchecked") + List result = (List) payload; + assertEquals(25, result.size()); + result.forEach(s -> assertTrue(evens.contains(Integer.parseInt(s)))); + } + + @Test + public void testFileWritingFlow() throws Exception { + String payload = "Spring Integration"; + this.fileWritingInput.send(new GenericMessage<>(payload)); + Message receive = this.fileWritingResultChannel.receive(1000); + assertNotNull(receive); + assertThat(receive.getPayload(), instanceOf(File.class)); + File resultFile = (File) receive.getPayload(); + assertThat(resultFile.getAbsolutePath(), + endsWith(TestUtils.applySystemFileSeparator("fileWritingFlow/foo.sitest"))); + String fileContent = StreamUtils.copyToString(new FileInputStream(resultFile), Charset.defaultCharset()); + assertEquals(payload, fileContent); + } + + @MessagingGateway(defaultRequestChannel = "controlBus.input") + private static interface ControlBusGateway { + + void send(String command); + + } + + @Configuration + @EnableIntegration + @IntegrationComponentScan + public static class ContextConfiguration { + + @Bean + public IntegrationFlow controlBus() { + return f -> f.controlBus(); + } + + @Bean + public IntegrationFlow fileFlow1() { + return IntegrationFlows.from("fileFlow1Input") + .handleWithAdapter(h -> h.file(tmpDir).fileNameGenerator(message -> null) + , c -> c.id("fileWriting")) + .get(); + } + + @Bean + public IntegrationFlow tailFlow() { + return IntegrationFlows.from((MessageProducers p) -> p.tail(new File(tmpDir, "TailTest")) + .delay(500) + .end(false) + .id("tailer") + .autoStartup(false)) + .transform("hello "::concat) + .channel(MessageChannels.queue("tailChannel")) + .get(); + } + + @Bean + public IntegrationFlow fileReadingFlow() { + return IntegrationFlows + .from(s -> s.file(tmpDir).patternFilter("*.sitest"), + e -> e.poller(Pollers.fixedDelay(100))) + .transform(Transformers.fileToString()) + .aggregate(a -> a.correlationExpression("1") + .releaseStrategy(g -> g.size() == 25), null) + .channel(MessageChannels.queue("fileReadingResultChannel")) + .get(); + } + + @Bean + public IntegrationFlow fileWritingFlow() { + return IntegrationFlows.from("fileWritingInput") + .enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.sitest") + .header("directory", new File(tmpDir, "fileWritingFlow"))) + .handleWithAdapter(a -> a.fileGateway(m -> m.getHeaders().get("directory"))) + .channel(MessageChannels.queue("fileWritingResultChannel")) + .get(); + } + + } + +} diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java index 5cf2fba..c9b5f0b 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java @@ -17,9 +17,7 @@ package org.springframework.integration.dsl.test.flows; 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; @@ -29,11 +27,6 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -import java.io.File; -import java.io.FileInputStream; -import java.io.FileOutputStream; -import java.io.IOException; -import java.nio.charset.Charset; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -43,39 +36,19 @@ import java.util.Map; import java.util.concurrent.Executor; 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 javax.management.MBeanServer; -import javax.management.MalformedObjectNameException; -import javax.management.ObjectName; 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; -import org.springframework.amqp.core.AmqpTemplate; -import org.springframework.amqp.core.AnonymousQueue; -import org.springframework.amqp.core.Queue; -import org.springframework.amqp.rabbit.connection.ConnectionFactory; -import org.springframework.aop.TargetSource; -import org.springframework.aop.framework.Advised; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.ConfigFileApplicationContextInitializer; import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext; @@ -83,9 +56,6 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.DependsOn; -import org.springframework.context.annotation.Import; -import org.springframework.data.mongodb.MongoDbFactory; -import org.springframework.data.mongodb.core.SimpleMongoDbFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.MessageRejectedException; @@ -94,51 +64,25 @@ import org.springframework.integration.annotation.IntegrationComponentScan; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.MessagingGateway; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.channel.ChannelInterceptorAware; import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; -import org.springframework.integration.config.GlobalChannelInterceptor; import org.springframework.integration.context.IntegrationContextUtils; -import org.springframework.integration.core.MessageSource; -import org.springframework.integration.dsl.Channels; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; -import org.springframework.integration.dsl.MessageProducers; -import org.springframework.integration.dsl.MessageSources; -import org.springframework.integration.dsl.MessagingGateways; -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.core.Pollers; -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.Transformers; -import org.springframework.integration.dsl.test.TestFtpServer; -import org.springframework.integration.dsl.test.TestSftpServer; -import org.springframework.integration.endpoint.MethodInvokingMessageSource; 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.FileWritingMessageHandler; -import org.springframework.integration.file.remote.RemoteFileTemplate; -import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; -import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer; -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; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.MutableMessageBuilder; -import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.transformer.PayloadDeserializingTransformer; import org.springframework.integration.transformer.PayloadSerializingTransformer; import org.springframework.integration.xml.transformer.support.XPathExpressionEvaluatingHeaderValueMessageProcessor; @@ -146,14 +90,11 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.core.DestinationResolutionException; -import org.springframework.messaging.simp.SimpMessageHeaderAccessor; -import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; @@ -163,49 +104,30 @@ import org.springframework.stereotype.Service; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import org.springframework.util.StreamUtils; - -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; /** * @author Artem Bilan * @author Tim Ysewyn */ -@ContextConfiguration(initializers = ConfigFileApplicationContextInitializer.class) +@ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext public class IntegrationFlowTests { - private static final File tmpDir = new File(System.getProperty("java.io.tmpdir")); - - private static int mongoPort; - - private static MongodExecutable mongodExe; - @Autowired private ListableBeanFactory beanFactory; @Autowired private ControlBusGateway controlBus; - @Autowired - @Qualifier("flow1QueueChannel") - private PollableChannel outputChannel; - - @Autowired - private TestChannelInterceptor testChannelInterceptor; - @Autowired @Qualifier("inputChannel") private MessageChannel inputChannel; + @Autowired + @Qualifier("discardChannel") + private PollableChannel discardChannel; + @Autowired @Qualifier("foo") private SubscribableChannel foo; @@ -237,14 +159,6 @@ public class IntegrationFlowTests { @Qualifier("bridgeFlow2Output") private PollableChannel bridgeFlow2Output; - @Autowired - @Qualifier("fileFlow1Input") - private MessageChannel fileFlow1Input; - - @Autowired - @Qualifier("fileWriting.handler") - private MessageHandler fileWritingMessageHandler; - @Autowired @Qualifier("methodInvokingInput") private MessageChannel methodInvokingInput; @@ -328,32 +242,10 @@ public class IntegrationFlowTests { @Qualifier("claimCheckInput") private MessageChannel claimCheckInput; - @Autowired - @Qualifier("priorityChannel") - private MessageChannel priorityChannel; - - @Autowired - @Qualifier("priorityReplyChannel") - private PollableChannel priorityReplyChannel; - @Autowired @Qualifier("lamdasInput") private MessageChannel lamdasInput; - @Autowired - @Qualifier("tailChannel") - private PollableChannel tailChannel; - - @Autowired - private ApacheCommonsFileTailingMessageProducer tailer; - - @Autowired - private AmqpTemplate amqpTemplate; - - @Autowired - @Qualifier("queue") - private Queue amqpQueue; - @Autowired @Qualifier("gatewayInput") private MessageChannel gatewayInput; @@ -362,39 +254,6 @@ public class IntegrationFlowTests { @Qualifier("gatewayError") private PollableChannel gatewayError; - @BeforeClass - public static void setup() throws IOException { - mongoPort = Network.getFreeServerPort(); - mongodExe = MongodStarter.getDefaultInstance() - .prepare(new MongodConfigBuilder() - .version(Version.Main.PRODUCTION) - .net(new Net(mongoPort, Network.localhostIsIPv6())) - .build()); - mongodExe.start(); - } - - @AfterClass - public static void tearDown() { - mongodExe.stop(); - } - - @Test - public void testPollingFlow() { - this.controlBus.send("@integerEndpoint.start()"); - assertThat(this.beanFactory.getBean("integerChannel"), instanceOf(FixedSubscriberChannel.class)); - for (int i = 0; i < 5; i++) { - Message message = this.outputChannel.receive(20000); - assertNotNull(message); - assertEquals("" + i, message.getPayload()); - } - this.controlBus.send("@integerEndpoint.stop()"); - - assertTrue(((ChannelInterceptorAware) this.outputChannel).getChannelInterceptors() - .contains(this.testChannelInterceptor)); - assertThat(this.testChannelInterceptor.getInvoked(), Matchers.greaterThanOrEqualTo(5)); - - } - @Test public void testDirectFlow() { assertTrue(this.beanFactory.containsBean("filter")); @@ -426,6 +285,11 @@ public class IntegrationFlowTests { assertEquals(100, successMessage.getPayload()); assertTrue(used.get()); + + this.inputChannel.send(new GenericMessage(1000)); + Message discarded = this.discardChannel.receive(5000); + assertNotNull(discarded); + assertEquals("Discarded: 1000", discarded.getPayload()); } @Test @@ -499,34 +363,6 @@ public class IntegrationFlowTests { } } - - @Test - public void testFileHandler() throws Exception { - Message message = MessageBuilder.withPayload("foo").setHeader(FileHeaders.FILENAME, "foo").build(); - try { - this.fileFlow1Input.send(message); - fail("NullPointerException expected"); - } - catch (Exception e) { - assertThat(e, instanceOf(MessageHandlingException.class)); - assertThat(e.getCause(), instanceOf(NullPointerException.class)); - } - DefaultFileNameGenerator fileNameGenerator = new DefaultFileNameGenerator(); - fileNameGenerator.setBeanFactory(this.beanFactory); - Object targetFileWritingMessageHandler = this.fileWritingMessageHandler; - if (this.fileWritingMessageHandler instanceof Advised) { - TargetSource targetSource = ((Advised) this.fileWritingMessageHandler).getTargetSource(); - if (targetSource != null) { - targetFileWritingMessageHandler = targetSource.getTarget(); - } - } - DirectFieldAccessor dfa = new DirectFieldAccessor(targetFileWritingMessageHandler); - dfa.setPropertyValue("fileNameGenerator", fileNameGenerator); - this.fileFlow1Input.send(message); - - assertTrue(new File(tmpDir, "foo").exists()); - } - @Test public void testMethodInvokingMessageHandler() { QueueChannel replyChannel = new QueueChannel(); @@ -702,6 +538,7 @@ public class IntegrationFlowTests { @Test public void testRouter() { + this.beanFactory.containsBean("routeFlow.subFlow#0.channel#0"); int[] payloads = new int[] {1, 2, 3, 4, 5, 6}; @@ -712,11 +549,11 @@ public class IntegrationFlowTests { for (int i = 0; i < 3; i++) { Message receive = this.oddChannel.receive(2000); assertNotNull(receive); - assertEquals(i * 2 + 1, receive.getPayload()); + assertEquals(payloads[i * 2] * 3, receive.getPayload()); receive = this.evenChannel.receive(2000); assertNotNull(receive); - assertEquals(i * 2 + 2, receive.getPayload()); + assertEquals(payloads[i * 2 + 1], receive.getPayload()); } } @@ -892,87 +729,6 @@ public class IntegrationFlowTests { assertSame(message, this.messageStore.getMessage(message.getHeaders().getId())); } - @Test - public void testPriority() throws InterruptedException { - Message message = MessageBuilder.withPayload("1").setPriority(1).build(); - this.priorityChannel.send(message); - - message = MessageBuilder.withPayload("-1").setPriority(-1).build(); - this.priorityChannel.send(message); - - message = MessageBuilder.withPayload("3").setPriority(3).build(); - this.priorityChannel.send(message); - - message = MessageBuilder.withPayload("0").setPriority(0).build(); - this.priorityChannel.send(message); - - message = MessageBuilder.withPayload("2").setPriority(2).build(); - this.priorityChannel.send(message); - - message = MessageBuilder.withPayload("none").build(); - this.priorityChannel.send(message); - - message = MessageBuilder.withPayload("31").setPriority(3).build(); - this.priorityChannel.send(message); - - this.controlBus.send("@priorityChannelBridge.start()"); - - Message receive = this.priorityReplyChannel.receive(2000); - assertNotNull(receive); - assertEquals("3", receive.getPayload()); - - receive = this.priorityReplyChannel.receive(2000); - assertNotNull(receive); - assertEquals("31", receive.getPayload()); - - receive = this.priorityReplyChannel.receive(2000); - assertNotNull(receive); - assertEquals("2", receive.getPayload()); - - receive = this.priorityReplyChannel.receive(2000); - assertNotNull(receive); - assertEquals("1", receive.getPayload()); - - receive = this.priorityReplyChannel.receive(2000); - assertNotNull(receive); - assertEquals("0", receive.getPayload()); - - receive = this.priorityReplyChannel.receive(2000); - assertNotNull(receive); - assertEquals("-1", receive.getPayload()); - - receive = this.priorityReplyChannel.receive(2000); - assertNotNull(receive); - assertEquals("none", receive.getPayload()); - - this.controlBus.send("@priorityChannelBridge.stop()"); - } - - @Test - public void testMessageProducerFlow() throws Exception { - FileOutputStream file = new FileOutputStream(new File(tmpDir, "TailTest")); - for (int i = 0; i < 50; i++) { - file.write((i + "\n").getBytes()); - } - this.tailer.start(); - for (int i = 0; i < 50; i++) { - Message message = this.tailChannel.receive(5000); - assertNotNull(message); - assertEquals("hello " + i, message.getPayload()); - } - assertNull(this.tailChannel.receive(1)); - - this.controlBus.send("@tailer.stop()"); - file.close(); - } - - - @Test - public void testAmqpInboundGatewayFlow() throws Exception { - Object result = this.amqpTemplate.convertSendAndReceive(this.amqpQueue.getName(), "world"); - assertEquals("HELLO WORLD", result); - } - @Test public void testGatewayFlow() throws Exception { PollableChannel replyChannel = new QueueChannel(); @@ -1000,297 +756,35 @@ public class IntegrationFlowTests { } @Autowired - @Qualifier("amqpOutboundInput") - private MessageChannel amqpOutboundInput; + @Qualifier("subscribersFlow.input") + private MessageChannel subscribersFlowInput; @Autowired - @Qualifier("amqpReplyChannel.channel") - private PollableChannel amqpReplyChannel; + @Qualifier("subscriber1Results") + private PollableChannel subscriber1Results; + + @Autowired + @Qualifier("subscriber2Results") + private PollableChannel subscriber2Results; + + @Autowired + @Qualifier("subscriber3Results") + private PollableChannel subscriber3Results; @Test - public void testAmqpOutboundFlow() throws Exception { - this.amqpOutboundInput.send(MessageBuilder.withPayload("hello through the amqp") - .setHeader("routingKey", "foo") - .build()); - Message receive = null; - int i = 0; - do { - receive = this.amqpReplyChannel.receive(); - if (receive != null) { - break; - } - Thread.sleep(100); - i++; - } while (i < 10); + public void testSubscribersSubFlows() { + this.subscribersFlowInput.send(new GenericMessage(2)); - assertNotNull(receive); - assertEquals("HELLO THROUGH THE AMQP", receive.getPayload()); - } + Message receive1 = this.subscriber1Results.receive(5000); + assertNotNull(receive1); + assertEquals(1, receive1.getPayload()); - @Autowired - @Qualifier("jmsOutboundFlow.input") - private MessageChannel jmsOutboundInboundChannel; - - @Autowired - @Qualifier("jmsOutboundInboundReplyChannel") - private PollableChannel jmsOutboundInboundReplyChannel; - - @Test - public void testJmsOutboundInboundFlow() { - this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS") - .setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsInbound") - .build()); - - Message receive = this.jmsOutboundInboundReplyChannel.receive(5000); - - assertNotNull(receive); - assertEquals("HELLO THROUGH THE JMS", receive.getPayload()); - - this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS") - .setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsMessageDriver") - .build()); - - receive = this.jmsOutboundInboundReplyChannel.receive(5000); - - assertNotNull(receive); - assertEquals("hello through the jms", receive.getPayload()); - } - - @Autowired - @Qualifier("jmsOutboundGatewayFlow.input") - private MessageChannel jmsOutboundGatewayChannel; - - @Test - public void testJmsPipelineFlow() { - PollableChannel replyChannel = new QueueChannel(); - Message message = MessageBuilder.withPayload("hello through the jms pipeline") - .setReplyChannel(replyChannel) - .setHeader("destination", "jmsPipelineTest") - .build(); - this.jmsOutboundGatewayChannel.send(message); - - Message receive = replyChannel.receive(5000); - - assertNotNull(receive); - assertEquals("HELLO THROUGH THE JMS PIPELINE", receive.getPayload()); - } - - @Autowired - @Qualifier("fileReadingResultChannel") - private PollableChannel fileReadingResultChannel; - - @Test - public void testFileReadingFlow() throws Exception { - List evens = new ArrayList<>(25); - for (int i = 0; i < 50; i++) { - boolean even = i % 2 == 0; - String extension = even ? ".sitest" : ".foofile"; - if (even) { - evens.add(i); - } - FileOutputStream file = new FileOutputStream(new File(tmpDir, i + extension)); - file.write(("" + i).getBytes()); - file.flush(); - file.close(); - } - - Message message = fileReadingResultChannel.receive(10000); - assertNotNull(message); - Object payload = message.getPayload(); - assertThat(payload, instanceOf(List.class)); - @SuppressWarnings("unchecked") - List result = (List) payload; - assertEquals(25, result.size()); - result.forEach(s -> assertTrue(evens.contains(Integer.parseInt(s)))); - } - - - @Autowired - @Qualifier("fileWritingInput") - private MessageChannel fileWritingInput; - - @Autowired - @Qualifier("fileWritingResultChannel") - private PollableChannel fileWritingResultChannel; - - @Test - public void testFileWritingFlow() throws Exception { - String payload = "Spring Integration"; - this.fileWritingInput.send(new GenericMessage<>(payload)); - Message receive = this.fileWritingResultChannel.receive(1000); - assertNotNull(receive); - assertThat(receive.getPayload(), instanceOf(File.class)); - File resultFile = (File) receive.getPayload(); - assertThat(resultFile.getAbsolutePath(), - endsWith(TestUtils.applySystemFileSeparator("fileWritingFlow/foo.sitest"))); - String fileContent = StreamUtils.copyToString(new FileInputStream(resultFile), Charset.defaultCharset()); - assertEquals(payload, fileContent); - } - - - @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(File.class)); - File file = (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 = (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(File.class)); - File file = (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 = (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 (File file : localFiles) { - assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"), - Matchers.containsString(dir)); - } - assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(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 (File file : localFiles) { - assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"), - Matchers.containsString(dir)); - } - assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"), - Matchers.containsString(dir + "subSftpSource")); - } - - @Autowired - private MBeanServer mBeanServer; - - @Test - public void testMBeansForDSL() throws MalformedObjectNameException { - assertFalse(this.mBeanServer.queryMBeans(ObjectName.getInstance("org.springframework.integration:" + - "bean=anonymous,name=sftpMgetInputChannel,type=MessageHandler"), null).isEmpty()); - assertFalse(this.mBeanServer.queryMBeans(ObjectName.getInstance("org.springframework.integration:" + - "type=MessageHandler,name=ftpMgetInputChannel,bean=anonymous"), null).isEmpty()); + Message receive2 = this.subscriber2Results.receive(5000); + assertNotNull(receive2); + assertEquals(4, receive2.getPayload()); + Message receive3 = this.subscriber3Results.receive(5000); + assertNotNull(receive3); + assertEquals(6, receive3.getPayload()); } @MessagingGateway(defaultRequestChannel = "controlBus") @@ -1300,40 +794,15 @@ public class IntegrationFlowTests { } @Configuration - @Import({TestFtpServer.class, TestSftpServer.class}) - @EnableAutoConfiguration + @EnableIntegration @IntegrationComponentScan public static class ContextConfiguration { - @Bean - public MessageSource integerMessageSource() { - MethodInvokingMessageSource source = new MethodInvokingMessageSource(); - source.setObject(new AtomicInteger()); - source.setMethodName("getAndIncrement"); - return source; - } - @Bean public IntegrationFlow controlBusFlow() { return IntegrationFlows.from("controlBus").controlBus().get(); } - @Autowired - private javax.jms.ConnectionFactory jmsConnectionFactory; - - @Bean - public IntegrationFlow flow1() { - return IntegrationFlows.from(this.integerMessageSource(), - c -> c.poller(Pollers.fixedRate(100)) - .id("integerEndpoint") - .autoStartup(false)) - .fixedSubscriberChannel("integerChannel") - .transform("payload.toString()") - .channel(Jms.pollableChannel("flow1QueueChannel", this.jmsConnectionFactory) - .destination("flow1QueueChannel")) - .get(); - } - @Bean(name = PollerMetadata.DEFAULT_POLLER) public PollerMetadata poller() { return Pollers.fixedRate(500).maxMessagesPerPoll(1).get(); @@ -1356,148 +825,6 @@ public class IntegrationFlowTests { return MessageChannels.publishSubscribe().get(); } - - @Bean - public IntegrationFlow jmsOutboundFlow() { - return f -> f.handleWithAdapter(h -> h.jms(this.jmsConnectionFactory) - .destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)); - } - - @Bean - public MessageChannel jmsOutboundInboundReplyChannel() { - return MessageChannels.queue().get(); - } - - @Bean - public IntegrationFlow jmsInboundFlow() { - return IntegrationFlows - .from((MessageSources s) -> s.jms(this.jmsConnectionFactory).destination("jmsInbound")) - .transform(String::toUpperCase) - .channel(this.jmsOutboundInboundReplyChannel()) - .get(); - } - - @Bean - public IntegrationFlow jmsMessageDriverFlow() { - return IntegrationFlows - .from(Jms.messageDriverChannelAdapter(this.jmsConnectionFactory) - .destination("jmsMessageDriver")) - .transform(String::toLowerCase) - .channel(this.jmsOutboundInboundReplyChannel()) - .get(); - } - - @Bean - public IntegrationFlow jmsOutboundGatewayFlow() { - return f -> f.handleWithAdapter(a -> a.jmsGateway(this.jmsConnectionFactory) - .replyContainer() - .requestDestination("jmsPipelineTest")); - } - - @Bean - public IntegrationFlow jmsInboundGatewayFlow() { - return IntegrationFlows.from((MessagingGateways g) -> g.jms(this.jmsConnectionFactory) - .destination("jmsPipelineTest")) - .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(s -> s.ftp(this.ftpSessionFactory) - .preserveTimestamp(true) - .remoteDirectory("ftpSource") - .regexFilter(".*\\.txt$") - .localFilename(f -> f.toUpperCase() + ".a") - .localDirectory(this.ftpServer.getTargetLocalDirectory()), - e -> e.id("ftpInboundAdapter")) - .channel(MessageChannels.queue("ftpInboundResultChannel")) - .get(); - } - - @Bean - public IntegrationFlow sftpInboundFlow() { - return IntegrationFlows - .from(s -> s.sftp(this.sftpSessionFactory) - .preserveTimestamp(true) - .remoteDirectory("sftpSource") - .regexFilter(".*\\.txt$") - .localFilenameExpression("#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 MessageHandler ftpOutboundGateway() { - return Ftp.outboundGateway(this.ftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET, - "payload") - .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE) - .regexFileNameFilter("(subFtpSource|.*1.txt)") - .localDirectoryExpression("@ftpServer.targetLocalDirectoryName + #remoteDirectory") - .localFilenameExpression("#remoteFileName.replaceFirst('ftpSource', 'localTarget')") - .get(); - } - - @Bean - public IntegrationFlow ftpMGetFlow() { - return IntegrationFlows.from("ftpMgetInputChannel") - .handle(ftpOutboundGateway()) - .channel(remoteFileOutputChannel()) - .get(); - } - - @Bean - public IntegrationFlow sftpMGetFlow() { - return IntegrationFlows.from("sftpMgetInputChannel") - .handleWithAdapter(h -> - h.sftpGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET, - "payload") - .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE) - .regexFileNameFilter("(subSftpSource|.*1.txt)") - .localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory") - .localFilenameExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')")) - .channel(remoteFileOutputChannel()) - .get(); - } - @Bean public IntegrationFlow routerAsNonLastFlow() { return f -> f.route(p -> p, r -> r.resolutionRequired(false)) @@ -1530,7 +857,11 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow flow2() { return IntegrationFlows.from(this.inputChannel) - .filter(p -> p instanceof String, c -> c.id("filter")) + .filter(p -> p instanceof String, e -> e + .id("filter") + .discardFlow(df -> df + .transform(String.class, "Discarded: "::concat) + .channel(c -> c.queue("discardChannel")))) .channel("foo") .fixedSubscriberChannel() .transform(Integer::parseInt) @@ -1549,26 +880,17 @@ public class IntegrationFlowTests { } @Bean - public MongoDbFactory mongoDbFactory() throws Exception { - return new SimpleMongoDbFactory(new MongoClient("localhost", mongoPort), "local"); - } - - @Bean - public MongoDbChannelMessageStore mongoDbChannelMessageStore(MongoDbFactory mongoDbFactory) { - MongoDbChannelMessageStore mongoDbChannelMessageStore = new MongoDbChannelMessageStore(mongoDbFactory); - mongoDbChannelMessageStore.setPriorityEnabled(true); - return mongoDbChannelMessageStore; - } - - @Bean - public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) { - return IntegrationFlows.from((Channels c) -> - c.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup")) - .bridge(s -> s.poller(Pollers.fixedDelay(100)) - .autoStartup(false) - .id("priorityChannelBridge")) - .channel(MessageChannels.queue("priorityReplyChannel")) - .get(); + public IntegrationFlow subscribersFlow() { + return flow -> flow + .publishSubscribeChannel(Executors.newCachedThreadPool(), s -> s + .subscribe(f -> f + .handle((p, h) -> p / 2) + .channel(c -> c.queue("subscriber1Results"))) + .subscribe(f -> f + .handle((p, h) -> p * 2) + .channel(c -> c.queue("subscriber2Results")))) + .handle((p, h) -> p * 3) + .channel(c -> c.queue("subscriber3Results")); } } @@ -1607,6 +929,7 @@ public class IntegrationFlowTests { public void onApplicationEvent(MessagingEvent event) { eventHolder().set(event.getMessage().getPayload()); } + }; } @@ -1696,14 +1019,6 @@ public class IntegrationFlowTests { @Configuration public static class ContextConfiguration4 { - @Bean - public IntegrationFlow fileFlow1() { - return IntegrationFlows.from("fileFlow1Input") - .handleWithAdapter(h -> h.file(tmpDir).fileNameGenerator(message -> null) - , c -> c.id("fileWriting")) - .get(); - } - @Autowired @Qualifier("integrationFlowTests.GreetingService") private MessageHandler greetingService; @@ -1826,11 +1141,6 @@ public class IntegrationFlowTests { .get(); } - @Bean - public QueueChannel oddChannel() { - return new QueueChannel(); - } - @Bean public QueueChannel evenChannel() { return new QueueChannel(); @@ -1840,9 +1150,10 @@ public class IntegrationFlowTests { public IntegrationFlow routeFlow() { return IntegrationFlows.from("routerInput") .route(p -> p % 2 == 0, - m -> m.suffix("Channel") - .channelMapping("true", "even") - .channelMapping("false", "odd")) + m -> m.channelMapping("true", "evenChannel") + .subFlowMapping("false", f -> + f.handle((p, h) -> p * 3))) + .channel(c -> c.queue("oddChannel")) .get(); } @@ -1878,64 +1189,6 @@ public class IntegrationFlowTests { .get(); } - @Bean - public IntegrationFlow tailFlow() { - return IntegrationFlows.from((MessageProducers p) -> p.tail(new File(tmpDir, "TailTest")) - .delay(500) - .end(false) - .id("tailer") - .autoStartup(false)) - .transform("hello "::concat) - .channel(MessageChannels.queue("tailChannel")) - .get(); - } - - @Autowired - private ConnectionFactory rabbitConnectionFactory; - - @Autowired - private AmqpTemplate amqpTemplate; - - @Bean - public Queue queue() { - return new AnonymousQueue(); - } - - @Bean - public IntegrationFlow amqpFlow() { - return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue())) - .transform("hello "::concat) - .transform(String.class, String::toUpperCase) - .get(); - } - - @Bean - public IntegrationFlow amqpOutboundFlow() { - return IntegrationFlows.from(Amqp.channel("amqpOutboundInput", this.rabbitConnectionFactory)) - .handle(Amqp.outboundAdapter(this.amqpTemplate).routingKeyExpression("headers.routingKey")) - .get(); - } - - @Bean - public Queue fooQueue() { - return new Queue("foo"); - } - - @Bean - public Queue amqpReplyChannel() { - return new Queue("amqpReplyChannel"); - } - - @Bean - public IntegrationFlow amqpInboundFlow() { - return IntegrationFlows.from((MessageProducers p) -> p.amqp(this.rabbitConnectionFactory, fooQueue())) - .transform(String.class, String::toUpperCase) - .channel(Amqp.pollableChannel(this.rabbitConnectionFactory) - .queueName("amqpReplyChannel") - .channelTransacted(true)) - .get(); - } - @Bean @DependsOn("gatewayRequestFlow") public IntegrationFlow gatewayFlow() { @@ -1957,29 +1210,6 @@ public class IntegrationFlowTests { return MessageChannels.queue().get(); } - - @Bean - public IntegrationFlow fileReadingFlow() { - return IntegrationFlows - .from(s -> s.file(tmpDir).patternFilter("*.sitest"), - e -> e.poller(Pollers.fixedDelay(100))) - .transform(Transformers.fileToString()) - .aggregate(a -> a.correlationExpression("1") - .releaseStrategy(g -> g.size() == 25), null) - .channel(MessageChannels.queue("fileReadingResultChannel")) - .get(); - } - - @Bean - public IntegrationFlow fileWritingFlow() { - return IntegrationFlows.from("fileWritingInput") - .enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.sitest") - .header("directory", new File(tmpDir, "fileWritingFlow"))) - .handleWithAdapter(a -> a.fileGateway(m -> m.getHeaders().get("directory"))) - .channel(MessageChannels.queue("fileWritingResultChannel")) - .get(); - } - } private static class RoutingTestBean { @@ -2045,24 +1275,6 @@ public class IntegrationFlowTests { } - @Component - @GlobalChannelInterceptor(patterns = "flow1QueueChannel") - public static class TestChannelInterceptor extends ChannelInterceptorAdapter { - - private final AtomicInteger invoked = new AtomicInteger(); - - @Override - public Message preSend(Message message, MessageChannel channel) { - this.invoked.incrementAndGet(); - return message; - } - - public Integer getInvoked() { - return invoked.get(); - } - - } - private static class TestPojo { private String name; diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/ftp/FtpTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/ftp/FtpTests.java new file mode 100644 index 0000000..84d1d82 --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/ftp/FtpTests.java @@ -0,0 +1,259 @@ +/* + * 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.ftp; + +import static org.hamcrest.Matchers.containsString; +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; +import static org.junit.Assert.assertThat; + +import java.io.File; +import java.util.List; +import java.util.regex.Matcher; + +import javax.management.MBeanServer; +import javax.management.MalformedObjectNameException; +import javax.management.ObjectName; + +import org.apache.commons.net.ftp.FTPFile; +import org.hamcrest.Matchers; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.ConfigFileApplicationContextInitializer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.core.Pollers; +import org.springframework.integration.dsl.ftp.Ftp; +import org.springframework.integration.dsl.sftp.Sftp; +import org.springframework.integration.dsl.test.sftp.TestSftpServer; +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.scheduling.PollerMetadata; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +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 Artem Bilan + */ +@ContextConfiguration(initializers = ConfigFileApplicationContextInitializer.class) +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class FtpTests { + + @Autowired + private ControlBusGateway controlBus; + + @Autowired + private MBeanServer mBeanServer; + + @Autowired + private TestFtpServer ftpServer; + + @Autowired + private DefaultFtpSessionFactory ftpSessionFactory; + + @Autowired + @Qualifier("ftpInboundResultChannel") + private PollableChannel ftpInboundResultChannel; + + @Autowired + @Qualifier("toFtpChannel") + private MessageChannel toFtpChannel; + + @Autowired + @Qualifier("ftpMgetInputChannel") + private MessageChannel ftpMgetInputChannel; + + @Autowired + @Qualifier("remoteFileOutputChannel") + private PollableChannel remoteFileOutputChannel; + + @Before + @After + public void setupRemoteFileServers() { + this.ftpServer.recursiveDelete(this.ftpServer.getTargetLocalDirectory()); + this.ftpServer.recursiveDelete(this.ftpServer.getTargetFtpDirectory()); + } + + @Test + public void testFtpInboundFlow() { + this.controlBus.send("@ftpInboundAdapter.start()"); + + Message message = this.ftpInboundResultChannel.receive(1000); + assertNotNull(message); + Object payload = message.getPayload(); + assertThat(payload, instanceOf(File.class)); + File file = (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 = (File) message.getPayload(); + assertThat(file.getName(), isOneOf("FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a")); + assertThat(file.getAbsolutePath(), containsString("ftpTest")); + + this.controlBus.send("@ftpInboundAdapter.stop()"); + } + + @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()); + } + + @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 (File file : localFiles) { + assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"), + Matchers.containsString(dir)); + } + assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"), + Matchers.containsString(dir + "subFtpSource")); + } + + @Test + public void testMBeansForDSL() throws MalformedObjectNameException { + assertFalse(this.mBeanServer.queryMBeans(ObjectName.getInstance("org.springframework.integration:" + + "type=MessageHandler,name=ftpMgetInputChannel,bean=anonymous"), null).isEmpty()); + } + + + @MessagingGateway(defaultRequestChannel = "controlBus.input") + private static interface ControlBusGateway { + + void send(String command); + + } + + @Configuration + @Import(TestFtpServer.class) + @EnableAutoConfiguration + @IntegrationComponentScan + public static class ContextConfiguration { + + @Autowired + private TestFtpServer ftpServer; + + @Autowired + private DefaultFtpSessionFactory ftpSessionFactory; + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata poller() { + return Pollers.fixedRate(500).maxMessagesPerPoll(1).get(); + } + + @Bean + public IntegrationFlow controlBus() { + return f -> f.controlBus(); + } + + @Bean + public IntegrationFlow ftpInboundFlow() { + return IntegrationFlows + .from(s -> s.ftp(this.ftpSessionFactory) + .preserveTimestamp(true) + .remoteDirectory("ftpSource") + .regexFilter(".*\\.txt$") + .localFilename(f -> f.toUpperCase() + ".a") + .localDirectory(this.ftpServer.getTargetLocalDirectory()), + e -> e.id("ftpInboundAdapter").autoStartup(false)) + .channel(MessageChannels.queue("ftpInboundResultChannel")) + .get(); + } + + @Bean + public IntegrationFlow ftpOutboundFlow() { + return IntegrationFlows.from("toFtpChannel") + .handle(Ftp.outboundAdapter(this.ftpSessionFactory) + .useTemporaryFileName(false) + .fileNameExpression("headers['" + FileHeaders.FILENAME + "']") + .remoteDirectory(this.ftpServer.getTargetFtpDirectory().getName()) + ).get(); + } + + @Bean + public PollableChannel remoteFileOutputChannel() { + return new QueueChannel(); + } + + @Bean + public MessageHandler ftpOutboundGateway() { + return Ftp.outboundGateway(this.ftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET, + "payload") + .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE) + .regexFileNameFilter("(subFtpSource|.*1.txt)") + .localDirectoryExpression("@ftpServer.targetLocalDirectoryName + #remoteDirectory") + .localFilenameExpression("#remoteFileName.replaceFirst('ftpSource', 'localTarget')") + .get(); + } + + @Bean + public IntegrationFlow ftpMGetFlow() { + return IntegrationFlows.from("ftpMgetInputChannel") + .handle(ftpOutboundGateway()) + .channel(remoteFileOutputChannel()) + .get(); + } + + } + +} 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/ftp/TestFtpServer.java similarity index 99% rename from spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestFtpServer.java rename to spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/ftp/TestFtpServer.java index fcac1dc..d57eb83 100644 --- 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/ftp/TestFtpServer.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dsl.test; +package org.springframework.integration.dsl.test.ftp; import java.io.File; import java.io.FileOutputStream; diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jdbc/JdbcTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jdbc/JdbcTests.java index b63537b..df4b723 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jdbc/JdbcTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jdbc/JdbcTests.java @@ -83,7 +83,7 @@ public class JdbcTests { new ResultSetIterator(ps.executeQuery(), (rs1, rowNum) -> new Foo(rs1.getInt(1), rs1.getString(2)))) - , null) + , e -> e.applySequence(false)) .channel(c -> c.queue("splitResultsChannel")); } @@ -116,7 +116,11 @@ public class JdbcTests { @Override public boolean hasNext() { try { - return !this.rs.isLast(); + boolean hasNext = !this.rs.isLast(); + if (!hasNext) { + this.rs.close(); + } + return hasNext; } catch (SQLException e) { throw new InvalidResultSetAccessException(e); diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jms/JmsTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jms/JmsTests.java new file mode 100644 index 0000000..c929010 --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jms/JmsTests.java @@ -0,0 +1,263 @@ +/* + * 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.jms; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.atomic.AtomicInteger; + +import javax.jms.ConnectionFactory; + +import org.hamcrest.Matchers; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.channel.ChannelInterceptorAware; +import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.MessageSources; +import org.springframework.integration.dsl.MessagingGateways; +import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.core.Pollers; +import org.springframework.integration.dsl.jms.Jms; +import org.springframework.integration.endpoint.MethodInvokingMessageSource; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.simp.SimpMessageHeaderAccessor; +import org.springframework.messaging.support.ChannelInterceptorAdapter; +import org.springframework.stereotype.Component; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class JmsTests { + + @Autowired + private ListableBeanFactory beanFactory; + + @Autowired + private ControlBusGateway controlBus; + + @Autowired + @Qualifier("flow1QueueChannel") + private PollableChannel outputChannel; + + @Autowired + @Qualifier("jmsOutboundFlow.input") + private MessageChannel jmsOutboundInboundChannel; + + @Autowired + @Qualifier("jmsOutboundInboundReplyChannel") + private PollableChannel jmsOutboundInboundReplyChannel; + + @Autowired + @Qualifier("jmsOutboundGatewayFlow.input") + private MessageChannel jmsOutboundGatewayChannel; + + @Autowired + private TestChannelInterceptor testChannelInterceptor; + + @Test + public void testPollingFlow() { + this.controlBus.send("@integerEndpoint.start()"); + assertThat(this.beanFactory.getBean("integerChannel"), instanceOf(FixedSubscriberChannel.class)); + for (int i = 0; i < 5; i++) { + Message message = this.outputChannel.receive(20000); + assertNotNull(message); + assertEquals("" + i, message.getPayload()); + } + this.controlBus.send("@integerEndpoint.stop()"); + + assertTrue(((ChannelInterceptorAware) this.outputChannel).getChannelInterceptors() + .contains(this.testChannelInterceptor)); + assertThat(this.testChannelInterceptor.invoked.get(), Matchers.greaterThanOrEqualTo(5)); + + } + + @Test + public void testJmsOutboundInboundFlow() { + this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS") + .setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsInbound") + .build()); + + Message receive = this.jmsOutboundInboundReplyChannel.receive(5000); + + assertNotNull(receive); + assertEquals("HELLO THROUGH THE JMS", receive.getPayload()); + + this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS") + .setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsMessageDriver") + .build()); + + receive = this.jmsOutboundInboundReplyChannel.receive(5000); + + assertNotNull(receive); + assertEquals("hello through the jms", receive.getPayload()); + } + + @Test + public void testJmsPipelineFlow() { + PollableChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("hello through the jms pipeline") + .setReplyChannel(replyChannel) + .setHeader("destination", "jmsPipelineTest") + .build(); + this.jmsOutboundGatewayChannel.send(message); + + Message receive = replyChannel.receive(5000); + + assertNotNull(receive); + assertEquals("HELLO THROUGH THE JMS PIPELINE", receive.getPayload()); + } + + @MessagingGateway(defaultRequestChannel = "controlBus.input") + private static interface ControlBusGateway { + + void send(String command); + + } + + @Configuration + @EnableAutoConfiguration + @IntegrationComponentScan + @ComponentScan + public static class ContextConfiguration { + + @Autowired + private ConnectionFactory jmsConnectionFactory; + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata poller() { + return Pollers.fixedRate(500).maxMessagesPerPoll(1).get(); + } + + @Bean + public IntegrationFlow controlBus() { + return f -> f.controlBus(); + } + + @Bean + public MessageSource integerMessageSource() { + MethodInvokingMessageSource source = new MethodInvokingMessageSource(); + source.setObject(new AtomicInteger()); + source.setMethodName("getAndIncrement"); + return source; + } + + @Bean + public IntegrationFlow flow1() { + return IntegrationFlows.from(integerMessageSource(), + c -> c.poller(p -> p.fixedRate(100)) + .id("integerEndpoint") + .autoStartup(false)) + .fixedSubscriberChannel("integerChannel") + .transform("payload.toString()") + .channel(Jms.pollableChannel("flow1QueueChannel", this.jmsConnectionFactory) + .destination("flow1QueueChannel")) + .get(); + } + + @Bean + public IntegrationFlow jmsOutboundFlow() { + return f -> f.handleWithAdapter(h -> h.jms(this.jmsConnectionFactory) + .destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)); + } + + @Bean + public MessageChannel jmsOutboundInboundReplyChannel() { + return MessageChannels.queue().get(); + } + + @Bean + public IntegrationFlow jmsInboundFlow() { + return IntegrationFlows + .from((MessageSources s) -> s.jms(this.jmsConnectionFactory).destination("jmsInbound")) + .transform(String::toUpperCase) + .channel(this.jmsOutboundInboundReplyChannel()) + .get(); + } + + @Bean + public IntegrationFlow jmsMessageDriverFlow() { + return IntegrationFlows + .from(Jms.messageDriverChannelAdapter(this.jmsConnectionFactory) + .destination("jmsMessageDriver")) + .transform(String::toLowerCase) + .channel(this.jmsOutboundInboundReplyChannel()) + .get(); + } + + @Bean + public IntegrationFlow jmsOutboundGatewayFlow() { + return f -> f.handleWithAdapter(a -> + a.jmsGateway(this.jmsConnectionFactory) + .replyContainer() + .requestDestination("jmsPipelineTest")); + } + + @Bean + public IntegrationFlow jmsInboundGatewayFlow() { + return IntegrationFlows.from((MessagingGateways g) -> + g.jms(this.jmsConnectionFactory) + .destination("jmsPipelineTest")) + .transform(String::toUpperCase) + .get(); + } + + } + + @Component + @GlobalChannelInterceptor(patterns = "flow1QueueChannel") + public static class TestChannelInterceptor extends ChannelInterceptorAdapter { + + private final AtomicInteger invoked = new AtomicInteger(); + + @Override + public Message preSend(Message message, MessageChannel channel) { + this.invoked.incrementAndGet(); + return message; + } + + } + +} diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mongodb/MongoDbTest.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mongodb/MongoDbTest.java new file mode 100644 index 0000000..5ac0bb2 --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mongodb/MongoDbTest.java @@ -0,0 +1,199 @@ +/* + * 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.mongodb; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.io.IOException; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +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.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.mongodb.MongoDbFactory; +import org.springframework.data.mongodb.core.SimpleMongoDbFactory; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.Channels; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.core.Pollers; +import org.springframework.integration.mongodb.store.MongoDbChannelMessageStore; +import org.springframework.integration.store.PriorityCapableChannelMessageStore; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +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 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class MongoDbTest { + + private static int mongoPort; + + private static MongodExecutable mongodExe; + + @BeforeClass + public static void setup() throws IOException { + mongoPort = Network.getFreeServerPort(); + mongodExe = MongodStarter.getDefaultInstance() + .prepare(new MongodConfigBuilder() + .version(Version.Main.PRODUCTION) + .net(new Net(mongoPort, Network.localhostIsIPv6())) + .build()); + mongodExe.start(); + } + + @AfterClass + public static void tearDown() { + mongodExe.stop(); + } + + @Autowired + private ControlBusGateway controlBus; + + @Autowired + @Qualifier("priorityChannel") + private MessageChannel priorityChannel; + + @Autowired + @Qualifier("priorityReplyChannel") + private PollableChannel priorityReplyChannel; + + + @Test + public void testPriority() throws InterruptedException { + Message message = MessageBuilder.withPayload("1").setPriority(1).build(); + this.priorityChannel.send(message); + + message = MessageBuilder.withPayload("-1").setPriority(-1).build(); + this.priorityChannel.send(message); + + message = MessageBuilder.withPayload("3").setPriority(3).build(); + this.priorityChannel.send(message); + + message = MessageBuilder.withPayload("0").setPriority(0).build(); + this.priorityChannel.send(message); + + message = MessageBuilder.withPayload("2").setPriority(2).build(); + this.priorityChannel.send(message); + + message = MessageBuilder.withPayload("none").build(); + this.priorityChannel.send(message); + + message = MessageBuilder.withPayload("31").setPriority(3).build(); + this.priorityChannel.send(message); + + this.controlBus.send("@priorityChannelBridge.start()"); + + Message receive = this.priorityReplyChannel.receive(2000); + assertNotNull(receive); + assertEquals("3", receive.getPayload()); + + receive = this.priorityReplyChannel.receive(2000); + assertNotNull(receive); + assertEquals("31", receive.getPayload()); + + receive = this.priorityReplyChannel.receive(2000); + assertNotNull(receive); + assertEquals("2", receive.getPayload()); + + receive = this.priorityReplyChannel.receive(2000); + assertNotNull(receive); + assertEquals("1", receive.getPayload()); + + receive = this.priorityReplyChannel.receive(2000); + assertNotNull(receive); + assertEquals("0", receive.getPayload()); + + receive = this.priorityReplyChannel.receive(2000); + assertNotNull(receive); + assertEquals("-1", receive.getPayload()); + + receive = this.priorityReplyChannel.receive(2000); + assertNotNull(receive); + assertEquals("none", receive.getPayload()); + + this.controlBus.send("@priorityChannelBridge.stop()"); + } + + @MessagingGateway(defaultRequestChannel = "controlBus.input") + private static interface ControlBusGateway { + + void send(String command); + } + + + @Configuration + @EnableIntegration + @IntegrationComponentScan + public static class ContextConfiguration { + + @Bean + public IntegrationFlow controlBus() { + return f -> f.controlBus(); + } + + @Bean + public MongoDbFactory mongoDbFactory() throws Exception { + return new SimpleMongoDbFactory(new MongoClient("localhost", mongoPort), "local"); + } + + @Bean + public MongoDbChannelMessageStore mongoDbChannelMessageStore(MongoDbFactory mongoDbFactory) { + MongoDbChannelMessageStore mongoDbChannelMessageStore = new MongoDbChannelMessageStore(mongoDbFactory); + mongoDbChannelMessageStore.setPriorityEnabled(true); + return mongoDbChannelMessageStore; + } + + @Bean + public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) { + return IntegrationFlows.from((Channels c) -> + c.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup")) + .bridge(s -> s.poller(Pollers.fixedDelay(100)) + .autoStartup(false) + .id("priorityChannelBridge")) + .channel(MessageChannels.queue("priorityReplyChannel")) + .get(); + } + + } + +} diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/sftp/SftpTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/sftp/SftpTests.java new file mode 100644 index 0000000..f7c632f --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/sftp/SftpTests.java @@ -0,0 +1,250 @@ +/* + * 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.sftp; + +import static org.hamcrest.Matchers.containsString; +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; +import static org.junit.Assert.assertThat; + +import java.io.File; +import java.util.List; +import java.util.regex.Matcher; + +import javax.management.MBeanServer; +import javax.management.MalformedObjectNameException; +import javax.management.ObjectName; + +import org.hamcrest.Matchers; +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.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.ConfigFileApplicationContextInitializer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.core.Pollers; +import org.springframework.integration.dsl.sftp.Sftp; +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.scheduling.PollerMetadata; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; +import org.springframework.integration.support.MessageBuilder; +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; + +import com.jcraft.jsch.ChannelSftp; + +/** + * @author Artem Bilan + */ +@ContextConfiguration(initializers = ConfigFileApplicationContextInitializer.class) +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class SftpTests { + + @Autowired + private ControlBusGateway controlBus; + + @Autowired + private MBeanServer mBeanServer; + + @Autowired + private TestSftpServer sftpServer; + + @Autowired + private DefaultSftpSessionFactory sftpSessionFactory; + + @Autowired + @Qualifier("sftpInboundResultChannel") + private PollableChannel sftpInboundResultChannel; + + @Autowired + @Qualifier("toSftpChannel") + private MessageChannel toSftpChannel; + + @Autowired + @Qualifier("remoteFileOutputChannel") + private PollableChannel remoteFileOutputChannel; + + + @Autowired + @Qualifier("sftpMgetInputChannel") + private MessageChannel sftpMgetInputChannel; + + @Before + @After + public void setupRemoteFileServers() { + this.sftpServer.recursiveDelete(this.sftpServer.getTargetLocalDirectory()); + this.sftpServer.recursiveDelete(this.sftpServer.getTargetSftpDirectory()); + } + + @Test + public void testSftpInboundFlow() { + this.controlBus.send("@sftpInboundAdapter.start()"); + + Message message = this.sftpInboundResultChannel.receive(1000); + assertNotNull(message); + Object payload = message.getPayload(); + assertThat(payload, instanceOf(File.class)); + File file = (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 = (File) message.getPayload(); + assertThat(file.getName(), isOneOf("SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a")); + assertThat(file.getAbsolutePath(), containsString("sftpTest")); + + this.controlBus.send("@sftpInboundAdapter.stop()"); + } + + @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()); + } + + @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 (File file : localFiles) { + assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"), + Matchers.containsString(dir)); + } + assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"), + Matchers.containsString(dir + "subSftpSource")); + } + + @Test + public void testMBeansForDSL() throws MalformedObjectNameException { + assertFalse(this.mBeanServer.queryMBeans(ObjectName.getInstance("org.springframework.integration:" + + "bean=anonymous,name=sftpMgetInputChannel,type=MessageHandler"), null).isEmpty()); + } + + + @MessagingGateway(defaultRequestChannel = "controlBus.input") + private static interface ControlBusGateway { + + void send(String command); + + } + + @Configuration + @Import(TestSftpServer.class) + @EnableAutoConfiguration + @IntegrationComponentScan + public static class ContextConfiguration { + + @Autowired + private TestSftpServer sftpServer; + + @Autowired + private DefaultSftpSessionFactory sftpSessionFactory; + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata poller() { + return Pollers.fixedRate(500).maxMessagesPerPoll(1).get(); + } + + @Bean + public IntegrationFlow controlBus() { + return f -> f.controlBus(); + } + + @Bean + public IntegrationFlow sftpInboundFlow() { + return IntegrationFlows + .from(s -> s.sftp(this.sftpSessionFactory) + .preserveTimestamp(true) + .remoteDirectory("sftpSource") + .regexFilter(".*\\.txt$") + .localFilenameExpression("#this.toUpperCase() + '.a'") + .localDirectory(this.sftpServer.getTargetLocalDirectory()), + e -> e.id("sftpInboundAdapter").autoStartup(false)) + .channel(MessageChannels.queue("sftpInboundResultChannel")) + .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 sftpMGetFlow() { + return IntegrationFlows.from("sftpMgetInputChannel") + .handleWithAdapter(h -> + h.sftpGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET, + "payload") + .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE) + .regexFileNameFilter("(subSftpSource|.*1.txt)") + .localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory") + .localFilenameExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')")) + .channel(remoteFileOutputChannel()) + .get(); + } + + } + +} 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/sftp/TestSftpServer.java similarity index 99% rename from spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestSftpServer.java rename to spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/sftp/TestSftpServer.java index 850bc9c..faf946d 100644 --- 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/sftp/TestSftpServer.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dsl.test; +package org.springframework.integration.dsl.test.sftp; import java.io.File; import java.io.FileOutputStream;