From feb610d92bc4d444fd36546c1b759f52a73a291f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 5 Mar 2014 20:36:51 +0200 Subject: [PATCH] Add `enrich()` and channel names option --- spring-integration-java-dsl/README.md | 13 +- spring-integration-java-dsl/build.gradle | 9 -- .../integration/dsl/EnricherSpec.java | 128 ++++++++++++++++++ .../dsl/IntegrationFlowBuilder.java | 91 ++++++++++--- .../integration/dsl/IntegrationFlows.java | 5 + ...slIntegrationConfigurationInitializer.java | 10 +- .../dsl/core/MessageChannelReference.java | 34 +++++ .../dsl/support/EnricherConfigurer.java | 29 ++++ .../dsl/test/IntegrationFlowTests.java | 100 +++++++++++++- 9 files changed, 375 insertions(+), 44 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EnricherSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessageChannelReference.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EnricherConfigurer.java diff --git a/spring-integration-java-dsl/README.md b/spring-integration-java-dsl/README.md index e34bfdf..809680f 100644 --- a/spring-integration-java-dsl/README.md +++ b/spring-integration-java-dsl/README.md @@ -26,15 +26,12 @@ Spring Integration message flows from Spring `@Configuration` classes. @Bean public IntegrationFlow myFlow() { - PollerMetadata pollerMetadata = new PollerMetadata(); - pollerMetadata.setTrigger(new PeriodicTrigger(100)); - - return IntegrationFlows.from(this.integerMessageSource()) - .poll(this.inputChannel(), pollerMetadata) - .transform((Integer p) -> p * 2) + return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100))) + .channel(this.inputChannel()) + .filter((Integer p) -> p > 0) .transform(Object::toString) - .channel(new QueueChannel()) - .build(); + .channel(MessageChannels.queue()) + .get(); } } ```` diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index e499f9f..3e921f1 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -24,10 +24,6 @@ compileTestJava { } ext { - junitVersion = '4.11' - log4jVersion = '1.2.17' - mockitoVersion = '1.9.5' - springVersion = '4.0.1.RELEASE' springIntegrationVersion = '4.0.0.BUILD-SNAPSHOT' linkHomepage = 'https://github.com/spring-projects/spring-integration-extensions' @@ -52,11 +48,6 @@ dependencies { compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion" testCompile "org.springframework.integration:spring-integration-test:$springIntegrationVersion" - testCompile "junit:junit-dep:$junitVersion" - testCompile "log4j:log4j:$log4jVersion" - testCompile "org.mockito:mockito-all:$mockitoVersion" - testCompile "org.springframework:spring-test:$springVersion" - testCompile "org.springframework.integration:spring-integration-event:$springIntegrationVersion" testCompile "org.springframework.integration:spring-integration-file:$springIntegrationVersion" diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EnricherSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EnricherSpec.java new file mode 100644 index 0000000..4ecd3bf --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EnricherSpec.java @@ -0,0 +1,128 @@ +package org.springframework.integration.dsl; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.dsl.core.IntegrationComponentSpec; +import org.springframework.integration.transformer.ContentEnricher; +import org.springframework.integration.transformer.support.AbstractHeaderValueMessageProcessor; +import org.springframework.integration.transformer.support.ExpressionEvaluatingHeaderValueMessageProcessor; +import org.springframework.integration.transformer.support.HeaderValueMessageProcessor; +import org.springframework.integration.transformer.support.StaticHeaderValueMessageProcessor; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + */ +public class EnricherSpec extends IntegrationComponentSpec { + + private final static SpelExpressionParser PARSER = new SpelExpressionParser(); + + private final ContentEnricher enricher = new ContentEnricher(); + + private final Map propertyExpressions = new HashMap(); + + private final Map> headerExpressions = new HashMap>(); + + EnricherSpec() { + } + + public EnricherSpec requestChannel(MessageChannel requestChannel) { + this.enricher.setRequestChannel(requestChannel); + return _this(); + } + + public EnricherSpec requestChannel(String requestChannel) { + this.enricher.setRequestChannelName(requestChannel); + return _this(); + } + + public EnricherSpec replyChannel(MessageChannel replyChannel) { + this.enricher.setReplyChannel(replyChannel); + return _this(); + } + + public EnricherSpec replyChannel(String replyChannel) { + this.enricher.setReplyChannelName(replyChannel); + return _this(); + } + + public EnricherSpec requestTimeout(Long requestTimeout) { + this.enricher.setRequestTimeout(requestTimeout); + return _this(); + } + + public EnricherSpec replyTimeout(Long replyTimeout) { + this.enricher.setReplyTimeout(replyTimeout); + return _this(); + } + + public EnricherSpec requestPayloadExpression(String requestPayloadExpression) { + this.enricher.setRequestPayloadExpression(PARSER.parseExpression(requestPayloadExpression)); + return _this(); + } + + public EnricherSpec shouldClonePayload(boolean shouldClonePayload) { + this.enricher.setShouldClonePayload(shouldClonePayload); + return _this(); + } + + public EnricherSpec property(String key, String value) { + this.propertyExpressions.put(key, new LiteralExpression(value)); + return _this(); + } + + public EnricherSpec propertyExpression(String key, String expression) { + Assert.notNull(key); + this.propertyExpressions.put(key, PARSER.parseExpression(expression)); + return _this(); + } + + public EnricherSpec header(String name, Object value) { + return this.header(name, value, null); + } + + public EnricherSpec header(String name, Object value, Boolean overwrite) { + AbstractHeaderValueMessageProcessor headerValueMessageProcessor = new StaticHeaderValueMessageProcessor(value); + headerValueMessageProcessor.setOverwrite(overwrite); + return this.header(name, headerValueMessageProcessor); + } + + public EnricherSpec headerExpression(String name, String expression) { + return this.headerExpression(name, expression, null, null); + } + + public EnricherSpec headerExpression(String name, String expression, Boolean overwrite) { + return this.headerExpression(name, expression, overwrite, null); + } + + public EnricherSpec headerExpression(String name, String expression, Class type) { + return this.headerExpression(name, expression, null, type); + } + + public EnricherSpec headerExpression(String name, String expression, Boolean overwrite, Class type) { + AbstractHeaderValueMessageProcessor headerValueMessageProcessor = + new ExpressionEvaluatingHeaderValueMessageProcessor(expression, type); + headerValueMessageProcessor.setOverwrite(overwrite); + return this.header(name, headerValueMessageProcessor); + } + + public EnricherSpec header(String name, HeaderValueMessageProcessor headerValueMessageProcessor) { + Assert.notNull(name); + this.headerExpressions.put(name, headerValueMessageProcessor); + return _this(); + } + + + @Override + protected ContentEnricher doGet() { + this.enricher.setPropertyExpressions(this.propertyExpressions); + this.enricher.setHeaderExpressions(this.headerExpressions); + return this.enricher; + } + +} 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 c4a14e3..a62b110 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 @@ -21,17 +21,20 @@ import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.core.GenericSelector; -import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessageSelector; import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.core.ConsumerEndpointSpec; +import org.springframework.integration.dsl.core.MessageChannelReference; import org.springframework.integration.dsl.support.BeanNameMethodInvokingMessageHandler; import org.springframework.integration.dsl.support.EndpointConfigurer; +import org.springframework.integration.dsl.support.EnricherConfigurer; import org.springframework.integration.filter.ExpressionEvaluatingSelector; import org.springframework.integration.filter.MessageFilter; import org.springframework.integration.filter.MethodInvokingSelector; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.handler.DelayHandler; +import org.springframework.integration.transformer.ContentEnricher; import org.springframework.integration.transformer.ExpressionEvaluatingTransformer; import org.springframework.integration.transformer.GenericTransformer; import org.springframework.integration.transformer.MessageTransformingHandler; @@ -68,15 +71,24 @@ public final class IntegrationFlowBuilder { return this; } + public IntegrationFlowBuilder channel(String messageChannelName) { + return this.channel(new MessageChannelReference(messageChannelName)); + } + public IntegrationFlowBuilder channel(MessageChannel messageChannel) { Assert.notNull(messageChannel); if (this.currentMessageChannel != null) { GenericEndpointSpec endpointSpec = new GenericEndpointSpec(new BridgeHandler()); - endpointSpec.get().getT1().setInputChannel(this.currentMessageChannel); + if (this.currentMessageChannel instanceof MessageChannelReference) { + endpointSpec.get().getT1().setInputChannelName(((MessageChannelReference) this.currentMessageChannel).getName()); + } + else { + endpointSpec.get().getT1().setInputChannel(this.currentMessageChannel); + } this.addComponent(endpointSpec).currentComponent(endpointSpec.get().getT2()); } this.currentMessageChannel = messageChannel; - return this.addComponent(this.currentMessageChannel).registerOutputChannelIfCan(this.currentMessageChannel); + return this.registerOutputChannelIfCan(this.currentMessageChannel); } public IntegrationFlowBuilder channel(MessageChannelSpec messageChannelSpec) { @@ -85,6 +97,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder transform(String expression) { + Assert.hasText(expression); return this.transform(new ExpressionEvaluatingTransformer(PARSER.parseExpression(expression))); } @@ -94,12 +107,14 @@ public final class IntegrationFlowBuilder { public IntegrationFlowBuilder transform(GenericTransformer genericTransformer, EndpointConfigurer> endpointConfigurer) { + Assert.notNull(genericTransformer); Transformer transformer = genericTransformer instanceof Transformer ? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer); return this.handle(new MessageTransformingHandler(transformer), endpointConfigurer); } public IntegrationFlowBuilder filter(String expression) { + Assert.hasText(expression); return this.filter(new ExpressionEvaluatingSelector(PARSER.parseExpression(expression))); } @@ -108,6 +123,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder filter(GenericSelector genericSelector, EndpointConfigurer endpointConfigurer) { + Assert.notNull(genericSelector); MessageSelector selector = genericSelector instanceof MessageSelector ? (MessageSelector) genericSelector : new MethodInvokingSelector(genericSelector); return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer); @@ -122,10 +138,11 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder handle(String beanName, String methodName, EndpointConfigurer> endpointConfigurer) { - return this.handle(new BeanNameMethodInvokingMessageHandler(beanName, methodName) , endpointConfigurer); + return this.handle(new BeanNameMethodInvokingMessageHandler(beanName, methodName), endpointConfigurer); } public IntegrationFlowBuilder handle(H messageHandler, EndpointConfigurer> endpointConfigurer) { + Assert.notNull(messageHandler); return this.register(new GenericEndpointSpec(messageHandler), endpointConfigurer); } @@ -145,22 +162,15 @@ public final class IntegrationFlowBuilder { return this.register(new GenericEndpointSpec(delayHandler), endpointConfigurer); } - private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) { - this.flow.addComponent(outputChannel); - if (this.currentComponent != null) { - if (this.currentComponent instanceof MessageProducer) { - ((MessageProducer) this.currentComponent).setOutputChannel(outputChannel); - } - else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) { - ((SourcePollingChannelAdapterFactoryBean) this.currentComponent).setOutputChannel(outputChannel); - } - else { - throw new BeanCreationException("The 'currentComponent' (" + this.currentComponent + ") is a one-way 'MessageHandler'" + - "and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow."); - } - this.currentComponent = null; - } - return this; + public IntegrationFlowBuilder enrich(EnricherConfigurer enricherConfigurer) { + return this.enrich(enricherConfigurer, null); + } + + public IntegrationFlowBuilder enrich(EnricherConfigurer enricherConfigurer, EndpointConfigurer> endpointConfigurer) { + Assert.notNull(enricherConfigurer); + EnricherSpec enricherSpec = new EnricherSpec(); + enricherConfigurer.configure(enricherSpec); + return this.register(new GenericEndpointSpec(enricherSpec.get()), endpointConfigurer); } private > IntegrationFlowBuilder register(S endpointSpec, EndpointConfigurer endpointConfigurer) { @@ -174,11 +184,50 @@ public final class IntegrationFlowBuilder { this.registerOutputChannelIfCan(inputChannel); } - endpointSpec.get().getT1().setInputChannel(inputChannel); + if (inputChannel instanceof MessageChannelReference) { + endpointSpec.get().getT1().setInputChannelName(((MessageChannelReference) inputChannel).getName()); + } + else { + endpointSpec.get().getT1().setInputChannel(inputChannel); + } return this.addComponent(endpointSpec).currentComponent(endpointSpec.get().getT2()); } + private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) { + this.flow.addComponent(outputChannel); + String channelName = null; + if (outputChannel instanceof MessageChannelReference) { + channelName = ((MessageChannelReference) outputChannel).getName(); + } + if (this.currentComponent != null) { + if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) { + AbstractReplyProducingMessageHandler messageProducer = (AbstractReplyProducingMessageHandler) this.currentComponent; + if (channelName != null) { + messageProducer.setOutputChannelName(channelName); + } + else { + messageProducer.setOutputChannel(outputChannel); + } + } + else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) { + SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = (SourcePollingChannelAdapterFactoryBean) this.currentComponent; + if (channelName != null) { + pollingChannelAdapterFactoryBean.setOutputChannelName(channelName); + } + else { + pollingChannelAdapterFactoryBean.setOutputChannel(outputChannel); + } + } + else { + throw new BeanCreationException("The 'currentComponent' (" + this.currentComponent + ") is a one-way 'MessageHandler'" + + "and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow."); + } + this.currentComponent = null; + } + return this; + } + public IntegrationFlow get() { return this.flow; } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index cd11ddc..315b4ee 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -19,6 +19,7 @@ package org.springframework.integration.dsl; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.channel.MessageChannelSpec; +import org.springframework.integration.dsl.core.MessageChannelReference; import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.messaging.MessageChannel; @@ -27,6 +28,10 @@ import org.springframework.messaging.MessageChannel; */ public final class IntegrationFlows { + public static IntegrationFlowBuilder from(String messageChannelName) { + return from(new MessageChannelReference(messageChannelName)); + } + public static IntegrationFlowBuilder from(MessageChannel messageChannel) { return new IntegrationFlowBuilder().channel(messageChannel); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java index a2fe70f..99d30c0 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java @@ -35,8 +35,8 @@ import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.GenericBeanDefinition; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.config.IntegrationConfigUtils; import org.springframework.integration.config.IntegrationConfigurationInitializer; -import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.config.InstanceBeanDefinition; import org.springframework.messaging.MessageHandler; @@ -103,7 +103,7 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig if (!messageHandlers.contains(messageHandler)) { String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); - String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null; + String[] handlerAlias = id != null ? new String[]{id + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX} : null; BeanComponentDefinition definitionHolder = new BeanComponentDefinition(new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias); BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry); @@ -115,6 +115,12 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig } registry.registerBeanDefinition(endpointBeanName, new InstanceBeanDefinition(endpoint)); } + else if (instance instanceof MessageChannelReference) { + String channelName = ((MessageChannelReference) instance).getName(); + if (!registry.containsBeanDefinition(channelName)) { + IntegrationConfigUtils.autoCreateDirectChannel(channelName, registry); + } + } else { String beanName = generateInstanceBeanDefinitionName(registry, instance); registry.registerBeanDefinition(beanName, component); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessageChannelReference.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessageChannelReference.java new file mode 100644 index 0000000..18d9b0e --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessageChannelReference.java @@ -0,0 +1,34 @@ +package org.springframework.integration.dsl.core; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public class MessageChannelReference implements MessageChannel { + + private final String name; + + public MessageChannelReference(String name) { + Assert.notNull(name); + this.name = name; + } + + public String getName() { + return name; + } + + @Override + public boolean send(Message message) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean send(Message message, long timeout) { + throw new UnsupportedOperationException(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EnricherConfigurer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EnricherConfigurer.java new file mode 100644 index 0000000..8f524ee --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EnricherConfigurer.java @@ -0,0 +1,29 @@ +/* + * 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.support; + +import org.springframework.integration.dsl.EnricherSpec; + +/** + * @author Artem Bilan + + */ +public interface EnricherConfigurer { + + void configure(EnricherSpec spec); + +} diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index 3679018..d032789 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -24,6 +24,9 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import java.io.File; +import java.util.Collections; +import java.util.Date; +import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -49,6 +52,7 @@ import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessageSource; @@ -101,6 +105,10 @@ public class IntegrationFlowTests { @Qualifier("inputChannel") private DirectChannel inputChannel; + @Autowired + @Qualifier("foo") + private PublishSubscribeChannel foo; + @Autowired @Qualifier("successChannel") private PollableChannel successChannel; @@ -144,6 +152,11 @@ public class IntegrationFlowTests { @Qualifier("delayedAdvice") private DelayedAdvice delayedAdvice; + @Autowired + @Qualifier("enricherInput") + private DirectChannel enricherInput; + + @Test public void testPollingFlow() { for (int i = 0; i < 10; i++) { @@ -169,6 +182,11 @@ public class IntegrationFlowTests { assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers")); } this.beanFactory.getBean("payloadSerializingTransformer", Lifecycle.class).start(); + + final AtomicBoolean used = new AtomicBoolean(); + + this.foo.subscribe(m -> used.set(true)); + this.inputChannel.send(message); Message reply = replyChannel.receive(5000); assertNotNull(reply); @@ -177,6 +195,8 @@ public class IntegrationFlowTests { Message successMessage = this.successChannel.receive(5000); assertNotNull(successMessage); assertEquals(100, successMessage.getPayload()); + + assertTrue(used.get()); } @Test @@ -266,6 +286,22 @@ public class IntegrationFlowTests { } } + @Test + public void testContentEnricher() { + QueueChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload(new TestPojo("Bar")).setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel).build(); + this.enricherInput.send(message); + Message receive = replyChannel.receive(5000); + assertNotNull(receive); + assertEquals("Bar Bar", receive.getHeaders().get("foo")); + Object payload = receive.getPayload(); + assertThat(payload, Matchers.instanceOf(TestPojo.class)); + TestPojo result = (TestPojo) payload; + assertEquals("Bar Bar", result.getName()); + assertNotNull(result.getDate()); + assertThat(new Date(), Matchers.greaterThan(result.getDate())); + } + @Configuration @EnableIntegration @@ -282,6 +318,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow flow1() { return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100))) + .channel("integerChannel") .transform("payload.toString()") .channel(MessageChannels.queue("flow1QueueChannel")) .get(); @@ -297,6 +334,11 @@ public class IntegrationFlowTests { return MessageChannels.direct().get(); } + @Bean + public PublishSubscribeChannel foo() { + return MessageChannels.publishSubscribe().get(); + } + } @Configuration @@ -324,6 +366,7 @@ public class IntegrationFlowTests { public IntegrationFlow flow2() { return IntegrationFlows.from(this.inputChannel) .filter(p -> p instanceof String, c -> c.id("filter")) + .channel("foo") .transform(Integer::parseInt) .transform(new PayloadSerializingTransformer(), c -> c.autoStartup(false).id("payloadSerializingTransformer")) @@ -375,7 +418,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow flow3() { - return IntegrationFlows.from(MessageChannels.direct("flow3Input")) + return IntegrationFlows.from("flow3Input") .handle(new ApplicationEventPublishingMessageHandler()) .get(); } @@ -389,7 +432,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow bridgeFlow2() { - return IntegrationFlows.from(MessageChannels.direct("bridgeFlow2Input")) + return IntegrationFlows.from("bridgeFlow2Input") .bridge(c -> c.autoStartup(false).id("bridge")) .delay("delayer", "200", c -> c.advice(this.delayedAdvice)) .channel(MessageChannels.queue("bridgeFlow2Output")) @@ -425,7 +468,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow fileFlow1() { - return IntegrationFlows.from(MessageChannels.direct("fileFlow1Input")) + return IntegrationFlows.from("fileFlow1Input") .handle(this.fileWritingMessageHandler(), c -> { FileWritingMessageHandler handler = c.get().getT2(); handler.setFileNameGenerator(message -> null); @@ -433,13 +476,34 @@ public class IntegrationFlowTests { }) .get(); } + @Bean public IntegrationFlow methodInvokingFlow() { - return IntegrationFlows.from(MessageChannels.direct("methodInvokingInput")) + return IntegrationFlows.from("methodInvokingInput") .handle("greetingService", null) .get(); } + @Bean + public IntegrationFlow enricherFlow() { + return IntegrationFlows.from("enricherInput") + .enrich(e -> e.requestChannel("enrichChannel") + .requestPayloadExpression("payload") + .shouldClonePayload(false) + .propertyExpression("name", "payload['name']") + .propertyExpression("date", "new java.util.Date()") + .headerExpression("foo", "payload['name']") + ) + .get(); + } + + @Bean + public IntegrationFlow enrichFlow() { + return IntegrationFlows.from("enrichChannel") + .>transform(p -> Collections.singletonMap("name", p.getName() + " Bar")) + .get(); + } + } @Component("greetingService") @@ -473,4 +537,32 @@ public class IntegrationFlowTests { } + private static class TestPojo { + + private String name; + + private Date date; + + private TestPojo(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public Date getDate() { + return date; + } + + public void setDate(Date date) { + this.date = date; + } + + } + }