From 3a98b1c974030d89f5e55af41a5d9a4801ad0f6f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 20 Mar 2017 17:27:04 -0400 Subject: [PATCH] Refactoring for some DSL methods To avoid some extra methods and let to operate with more cleaner API, merge some specs to single entity * Make `EnricherSpec` and `AbstractRouterSpec` as `extends ConsumerEndpointSpec` * Remove extra methods in the `IntegrationFlowDefinition` * Provide refactoring and fixed to `enrich()` and `route()` functions according the `ConsumerEndpointSpec` merge * Port `TransformerTests` from SI-Java-DSL project --- .../integration/dsl/AbstractRouterSpec.java | 23 +- .../integration/dsl/ConsumerEndpointSpec.java | 2 +- .../integration/dsl/EndpointSpec.java | 10 +- .../integration/dsl/EnricherSpec.java | 38 +- .../integration/dsl/FilterEndpointSpec.java | 2 +- .../dsl/IntegrationFlowDefinition.java | 277 ++------------ .../dsl/RecipientListRouterSpec.java | 12 +- .../integration/dsl/RouterSpec.java | 27 +- .../dsl/transformers/TransformerTests.java | 357 ++++++++++++++++++ 9 files changed, 454 insertions(+), 294 deletions(-) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AbstractRouterSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/AbstractRouterSpec.java index 20f4ba239e..6da90343af 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AbstractRouterSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/AbstractRouterSpec.java @@ -16,10 +16,6 @@ package org.springframework.integration.dsl; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; - import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.router.AbstractMessageRouter; import org.springframework.messaging.MessageChannel; @@ -36,14 +32,12 @@ import org.springframework.util.Assert; * @since 5.0 */ public class AbstractRouterSpec, R extends AbstractMessageRouter> - extends MessageHandlerSpec implements ComponentsRegistration { - - protected final List componentsToRegister = new ArrayList<>(); + extends ConsumerEndpointSpec { private boolean defaultToParentFlow; AbstractRouterSpec(R router) { - this.target = router; + super(router); } /** @@ -52,7 +46,7 @@ public class AbstractRouterSpec, R extends Ab * @see AbstractMessageRouter#setIgnoreSendFailures(boolean) */ public S ignoreSendFailures(boolean ignoreSendFailures) { - this.target.setIgnoreSendFailures(ignoreSendFailures); + this.handler.setIgnoreSendFailures(ignoreSendFailures); return _this(); } @@ -62,7 +56,7 @@ public class AbstractRouterSpec, R extends Ab * @see AbstractMessageRouter#setApplySequence(boolean) */ public S applySequence(boolean applySequence) { - this.target.setApplySequence(applySequence); + this.handler.setApplySequence(applySequence); return _this(); } @@ -74,7 +68,7 @@ public class AbstractRouterSpec, R extends Ab * @see AbstractMessageRouter#setDefaultOutputChannelName(String) */ public S defaultOutputChannel(String channelName) { - this.target.setDefaultOutputChannelName(channelName); + this.handler.setDefaultOutputChannelName(channelName); return _this(); } @@ -86,7 +80,7 @@ public class AbstractRouterSpec, R extends Ab * @see AbstractMessageRouter#setDefaultOutputChannel(MessageChannel) */ public S defaultOutputChannel(MessageChannel channel) { - this.target.setDefaultOutputChannel(channel); + this.handler.setDefaultOutputChannel(channel); return _this(); } @@ -123,9 +117,4 @@ public class AbstractRouterSpec, R extends Ab return this.defaultToParentFlow; } - @Override - public Collection getComponentsToRegister() { - return this.componentsToRegister; - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java index e3374d63f0..d894706608 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java @@ -150,7 +150,7 @@ public abstract class ConsumerEndpointSpec, */ public S transactional(boolean handleMessageAdvice) { TransactionInterceptor transactionInterceptor = new TransactionInterceptorBuilder(handleMessageAdvice).build(); - this.componentToRegister.add(transactionInterceptor); + this.componentsToRegister.add(transactionInterceptor); return transactional(transactionInterceptor); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java index e58ef57f8d..0fc0764f89 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -46,7 +46,7 @@ public abstract class EndpointSpec, F extends Be extends IntegrationComponentSpec> implements ComponentsRegistration { - protected final Collection componentToRegister = new ArrayList(); + protected final Collection componentsToRegister = new ArrayList<>(); protected H handler; @@ -89,7 +89,7 @@ public abstract class EndpointSpec, F extends Be public S poller(PollerSpec pollerMetadataSpec) { Collection componentsToRegister = pollerMetadataSpec.getComponentsToRegister(); if (componentsToRegister != null) { - this.componentToRegister.addAll(componentsToRegister); + this.componentsToRegister.addAll(componentsToRegister); } return poller(pollerMetadataSpec.get()); } @@ -117,9 +117,9 @@ public abstract class EndpointSpec, F extends Be @Override public Collection getComponentsToRegister() { - return this.componentToRegister.isEmpty() + return this.componentsToRegister.isEmpty() ? null - : this.componentToRegister; + : this.componentsToRegister; } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java index 41d3c8bd8d..75826951ec 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java @@ -21,6 +21,7 @@ import java.util.Map; import java.util.function.Function; import org.springframework.expression.Expression; +import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.transformer.ContentEnricher; @@ -32,17 +33,17 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; +import reactor.util.function.Tuple2; + /** - * The {@link MessageHandlerSpec} implementation for the {@link ContentEnricher}. + * A {@link ConsumerEndpointSpec} extension for the {@link ContentEnricher}. * * @author Artem Bilan * @author Tim Ysewyn * * @since 5.0 */ -public class EnricherSpec extends MessageHandlerSpec { - - private final ContentEnricher enricher = new ContentEnricher(); +public class EnricherSpec extends ConsumerEndpointSpec { private final Map propertyExpressions = new HashMap(); @@ -50,7 +51,7 @@ public class EnricherSpec extends MessageHandlerSpec>(); EnricherSpec() { - super(); + super(new ContentEnricher()); } /** @@ -59,7 +60,7 @@ public class EnricherSpec extends MessageHandlerSpec EnricherSpec requestPayload(Function, ?> requestPayloadFunction) { - this.enricher.setRequestPayloadExpression(new FunctionExpression<>(requestPayloadFunction)); + this.handler.setRequestPayloadExpression(new FunctionExpression<>(requestPayloadFunction)); return _this(); } @@ -141,7 +142,7 @@ public class EnricherSpec extends MessageHandlerSpec doGet() { if (!this.propertyExpressions.isEmpty()) { - this.enricher.setPropertyExpressions(this.propertyExpressions); + this.handler.setPropertyExpressions(this.propertyExpressions); } if (!this.headerExpressions.isEmpty()) { - this.enricher.setHeaderExpressions(this.headerExpressions); + this.handler.setHeaderExpressions(this.headerExpressions); } - return this.enricher; + return super.doGet(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java index d3c0519fb2..1f1562deae 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java @@ -91,7 +91,7 @@ public final class FilterEndpointSpec extends ConsumerEndpointSpec e.requestChannel("enrichChannel") * .requestPayload(Message::getPayload) * .shouldClonePayload(false) + * .autoStartup(false) * .>headerFunction("foo", m -> m.getPayload().get("name"))) * } * @@ -1172,35 +1172,7 @@ public abstract class IntegrationFlowDefinition enricherConfigurer) { - return this.enrich(enricherConfigurer, null); - } - - /** - * Populate a {@link ContentEnricher} to the current integration flow position - * with provided options. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. - * Typically used with a Java 8 Lambda expression: - * 
-	 * {@code
-	 *  .enrich(e -> e.requestChannel("enrichChannel")
-	 *                  .requestPayload(Message::getPayload)
-	 *                  .shouldClonePayload(false)
-	 *                  .>headerFunction("foo", m -> m.getPayload().get("name")),
-	 *           e -> e.autoStartup(false))
-	 * }
-	 *
- * @param enricherConfigurer the {@link Consumer} to provide {@link ContentEnricher} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @return the current {@link IntegrationFlowDefinition}. - * @see EnricherSpec - * @see GenericEndpointSpec - */ - public B enrich(Consumer enricherConfigurer, - Consumer> endpointConfigurer) { - Assert.notNull(enricherConfigurer, "'enricherConfigurer' must not be null"); - EnricherSpec enricherSpec = new EnricherSpec(); - enricherConfigurer.accept(enricherSpec); - return handle(enricherSpec.get(), endpointConfigurer); + return register(new EnricherSpec(), enricherConfigurer); } /** @@ -1272,14 +1244,7 @@ public abstract class IntegrationFlowDefinition headers, Consumer> endpointConfigurer) { - return enrichHeaders(new Consumer() { - - @Override - public void accept(HeaderEnricherSpec spec) { - spec.headers(headers); - } - - }, endpointConfigurer); + return enrichHeaders(spec -> spec.headers(headers), endpointConfigurer); } /** @@ -1806,7 +1771,7 @@ public abstract class IntegrationFlowDefinition> routerConfigurer) { - return route(beanName, method, routerConfigurer, null); - } - - /** - * Populate the {@link MethodInvokingRouter} for provided bean and its method - * with provided options from {@link RouterSpec}. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. - * @param beanName the bean to use. - * @param method the method to invoke at runtime. - * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @return the current {@link IntegrationFlowDefinition}. - */ - public B route(String beanName, String method, Consumer> routerConfigurer, - Consumer> endpointConfigurer) { - return this.route(new MethodInvokingRouter(new BeanNameMessageProcessor(beanName, method)), - routerConfigurer, endpointConfigurer); + public B route(String beanName, String method, Consumer> routerConfigurer) { + MethodInvokingRouter methodInvokingRouter = + new MethodInvokingRouter(new BeanNameMessageProcessor<>(beanName, method)); + return route(new RouterSpec<>(methodInvokingRouter), routerConfigurer); } /** @@ -1872,23 +1823,6 @@ public abstract class IntegrationFlowDefinition> routerConfigurer) { - return route(service, methodName, routerConfigurer, null); - } - - /** - * Populate the {@link MethodInvokingRouter} for the method - * of the provided service and its method with provided options from {@link RouterSpec}. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. - * @param service the service to use. - * @param methodName the method to invoke. - * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @return the current {@link IntegrationFlowDefinition}. - * @see MethodInvokingRouter - */ - public B route(Object service, String methodName, - Consumer> routerConfigurer, - Consumer> endpointConfigurer) { MethodInvokingRouter router; if (StringUtils.hasText(methodName)) { router = new MethodInvokingRouter(service, methodName); @@ -1896,7 +1830,7 @@ public abstract class IntegrationFlowDefinition(router), routerConfigurer); } @@ -1907,7 +1841,7 @@ public abstract class IntegrationFlowDefinition>) null); + return route(expression, (Consumer>) null); } /** @@ -1919,23 +1853,8 @@ public abstract class IntegrationFlowDefinition B route(String expression, Consumer> routerConfigurer) { - return route(expression, routerConfigurer, null); - } - - /** - * Populate the {@link ExpressionEvaluatingRouter} for provided bean and its method - * with provided options from {@link RouterSpec}. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. - * @param expression the expression to use. - * @param routerConfigurer the {@link Consumer} to provide {@link ExpressionEvaluatingRouter} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @param the target result type. - * @return the current {@link IntegrationFlowDefinition}. - */ - public B route(String expression, Consumer> routerConfigurer, - Consumer> endpointConfigurer) { - return this.route(new ExpressionEvaluatingRouter(PARSER.parseExpression(expression)), routerConfigurer, - endpointConfigurer); + return route(new RouterSpec<>(new ExpressionEvaluatingRouter(PARSER.parseExpression(expression))), + routerConfigurer); } /** @@ -1953,29 +1872,7 @@ public abstract class IntegrationFlowDefinition B route(Function router) { - return this.route(null, router); - } - - /** - * Populate the {@link MethodInvokingRouter} for provided {@link Function} - * with provided options from {@link RouterSpec}. - * Typically used with a Java 8 Lambda expression: - * 
-	 * {@code
-	 *  .route(p -> p % 2 == 0,
-	 *                 m -> m.channelMapping("true", "evenChannel")
-	 *                       .subFlowMapping("false", f ->
-	 *                                   f.handle((p, h) -> p * 3)))
-	 * }
-	 *
- * @param router the {@link Function} to use. - * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. - * @param the source payload type. - * @param the target result type. - * @return the current {@link IntegrationFlowDefinition}. - */ - public B route(Function router, Consumer> routerConfigurer) { - return this.route(null, router, routerConfigurer); + return route(null, router); } /** @@ -1995,32 +1892,7 @@ public abstract class IntegrationFlowDefinition B route(Class payloadType, Function router) { - return this.route(payloadType, router, null, null); - } - - /** - * Populate the {@link MethodInvokingRouter} for provided {@link Function} - * and payload type and options from {@link RouterSpec}. - * Typically used with a Java 8 Lambda expression: - * 
-	 * {@code
-	 *  .route(Integer.class, p -> p % 2 == 0,
-	 *                 m -> m.channelMapping("true", "evenChannel")
-	 *                       .subFlowMapping("false", f ->
-	 *                                   f.handle((p, h) -> p * 3)))
-	 * }
-	 *
- * @param payloadType the expected payload type. - * @param router the {@link Function} to use. - * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. - * @param the source payload type. - * @param the target result type. - * @return the current {@link IntegrationFlowDefinition}. - * @see LambdaMessageProcessor - */ - public B route(Class payloadType, Function router, - Consumer> routerConfigurer) { - return this.route(payloadType, router, routerConfigurer, null); + return route(payloadType, router, null); } /** @@ -2033,20 +1905,18 @@ public abstract class IntegrationFlowDefinitionroute(p -> p % 2 == 0, * m -> m.channelMapping("true", "evenChannel") * .subFlowMapping("false", f -> - * f.handle((p, h) -> p * 3)), - * e -> e.applySequence(false)) + * f.handle((p, h) -> p * 3)) + * .applySequence(false)) * } * * @param router the {@link Function} to use. * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. * @param the source payload type. * @param the target result type. * @return the current {@link IntegrationFlowDefinition}. */ - public B route(Function router, Consumer> routerConfigurer, - Consumer> endpointConfigurer) { - return route(null, router, routerConfigurer, endpointConfigurer); + public B route(Function router, Consumer> routerConfigurer) { + return route(null, router, routerConfigurer); } /** @@ -2059,26 +1929,24 @@ public abstract class IntegrationFlowDefinition p % 2 == 0, * m -> m.channelMapping("true", "evenChannel") * .subFlowMapping("false", f -> - * f.handle((p, h) -> p * 3)), - * e -> e.applySequence(false)) + * f.handle((p, h) -> p * 3)) + * .applySequence(false)) * } * * @param payloadType the expected payload type. * @param router the {@link Function} to use. * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. * @param

the source payload type. * @param the target result type. * @return the current {@link IntegrationFlowDefinition}. * @see LambdaMessageProcessor */ public B route(Class

payloadType, Function router, - Consumer> routerConfigurer, - Consumer> endpointConfigurer) { + Consumer> routerConfigurer) { MethodInvokingRouter methodInvokingRouter = isLambda(router) ? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType)) : new MethodInvokingRouter(router); - return route(methodInvokingRouter, routerConfigurer, endpointConfigurer); + return route(new RouterSpec<>(methodInvokingRouter), routerConfigurer); } /** @@ -2113,64 +1981,21 @@ public abstract class IntegrationFlowDefinition messageProcessorSpec, Consumer> routerConfigurer) { - return route(messageProcessorSpec, routerConfigurer, null); - } - - /** - * Populate the {@link MethodInvokingRouter} for the {@link org.springframework.integration.handler.MessageProcessor} - * from the provided {@link MessageProcessorSpec} with default options. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. - * 

-	 * {@code
-	 *  .route(Scripts.script(myScriptResource).lang("groovy").refreshCheckDelay(1000),
-	 *                 m -> m.channelMapping("true", "evenChannel")
-	 *                       .subFlowMapping("false", f ->
-	 *                                   f.handle((p, h) -> p * 3)),
-	 *                 e -> e.applySequence(false))
-	 * }
-	 *
- * @param messageProcessorSpec the {@link MessageProcessorSpec} to use. - * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @return the current {@link IntegrationFlowDefinition}. - */ - public B route(MessageProcessorSpec messageProcessorSpec, - Consumer> routerConfigurer, - Consumer> endpointConfigurer) { Assert.notNull(messageProcessorSpec, "'messageProcessorSpec' must not be null"); MessageProcessor processor = messageProcessorSpec.get(); - return addComponent(processor) - .route(new MethodInvokingRouter(processor), routerConfigurer, endpointConfigurer); + addComponent(processor); + + return route(new RouterSpec<>(new MethodInvokingRouter(processor)), routerConfigurer); } - /** - * Populate the provided {@link AbstractMappingMessageRouter} implementation - * with options from {@link RouterSpec} and endpoint options from {@link GenericEndpointSpec}. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. - * @param router the {@link AbstractMappingMessageRouter} to populate. - * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @param the {@code channelKey mapping} type. - * @param the {@link AbstractMappingMessageRouter} type. - * @return the current {@link IntegrationFlowDefinition}. - */ - public B route(R router, Consumer> routerConfigurer, - Consumer> endpointConfigurer) { + private > B route(S routerSpec, + Consumer routerConfigurer) { - RouterSpec routerSpec = new RouterSpec<>(router); if (routerConfigurer != null) { routerConfigurer.accept(routerSpec); } - return route(router, routerSpec, endpointConfigurer); - } - - private > B route(R router, - S routerSpec, Consumer> endpointConfigurer) { - - route(router, endpointConfigurer); - - final BridgeHandler bridgeHandler = new BridgeHandler(); + BridgeHandler bridgeHandler = new BridgeHandler(); boolean registerSubflowBridge = false; Collection componentsToRegister = routerSpec.getComponentsToRegister(); if (!CollectionUtils.isEmpty(componentsToRegister)) { @@ -2188,6 +2013,13 @@ public abstract class IntegrationFlowDefinition routerConfigurer) { - return routeToRecipients(routerConfigurer, null); - } - - /** - * Populate the {@link RecipientListRouter} options from {@link RecipientListRouterSpec}. - * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. - * Typically used with a Java 8 Lambda expression: - * 
-	 * {@code
-	 *  .routeToRecipients(r -> r
-	 *.recipient("bar-channel", m ->
-	 *            m.getHeaders().containsKey("recipient") && (boolean) m.getHeaders().get("recipient"))
-	 *      .recipientFlow("'foo' == payload or 'bar' == payload or 'baz' == payload",
-	 *                         f -> f.transform(String.class, p -> p.toUpperCase())
-	 *                               .channel(c -> c.queue("recipientListSubFlow1Result"))),
-	 *      e -> e.applySequence(false))
-	 * }
-	 *
- * @param routerConfigurer the {@link Consumer} to provide {@link RecipientListRouter} options. - * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @return the current {@link IntegrationFlowDefinition}. - */ - public B routeToRecipients(Consumer routerConfigurer, - Consumer> endpointConfigurer) { - - RecipientListRouterSpec spec = new RecipientListRouterSpec(); - if (routerConfigurer != null) { - routerConfigurer.accept(spec); - } - - return route(spec.get(), spec, endpointConfigurer); + return route(new RecipientListRouterSpec(), routerConfigurer); } /** @@ -2682,8 +2484,9 @@ public abstract class IntegrationFlowDefinition private String suffix; + private boolean mappingProviderRegistered; + RouterSpec(R router) { super(router); - this.mappingProvider = new RouterMappingProvider(this.target); + this.mappingProvider = new RouterMappingProvider(this.handler); } /** @@ -62,7 +64,7 @@ public final class RouterSpec * @see AbstractMappingMessageRouter#setResolutionRequired(boolean) */ public RouterSpec resolutionRequired(boolean resolutionRequired) { - this.target.setResolutionRequired(resolutionRequired); + this.handler.setResolutionRequired(resolutionRequired); return _this(); } @@ -73,9 +75,10 @@ public final class RouterSpec * @see AbstractMappingMessageRouter#setPrefix(String) */ public RouterSpec prefix(String prefix) { - Assert.state(this.componentsToRegister.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); + Assert.state(this.componentsToRegister.isEmpty(), + "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); this.prefix = prefix; - this.target.setPrefix(prefix); + this.handler.setPrefix(prefix); return _this(); } @@ -86,9 +89,10 @@ public final class RouterSpec * @see AbstractMappingMessageRouter#setSuffix(String) */ public RouterSpec suffix(String suffix) { - Assert.state(this.componentsToRegister.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); + Assert.state(this.componentsToRegister.isEmpty(), + "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive"); this.suffix = suffix; - this.target.setSuffix(suffix); + this.handler.setSuffix(suffix); return _this(); } @@ -102,7 +106,7 @@ public final class RouterSpec Assert.notNull(key, "'key' must not be null"); Assert.hasText(channelName, "'channelName' must not be null"); if (key instanceof String) { - this.target.setChannelMapping((String) key, channelName); + this.handler.setChannelMapping((String) key, channelName); } else { this.mappingProvider.addMapping(key, new NamedComponent() { @@ -148,9 +152,14 @@ public final class RouterSpec @Override public Collection getComponentsToRegister() { - // The 'mappingProvider' must be added to the 'componentToRegister' in the end to + // The 'mappingProvider' must be added to the 'componentsToRegister' in the end to // let all other components to be registered before the 'RouterMappingProvider.onInit()' logic. - this.componentsToRegister.add(this.mappingProvider); + if (!this.mappingProviderRegistered) { + if (!this.mappingProvider.mapping.isEmpty()) { + this.componentsToRegister.add(this.mappingProvider); + } + this.mappingProviderRegistered = true; + } return super.getComponentsToRegister(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java new file mode 100644 index 0000000000..2d602b4724 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java @@ -0,0 +1,357 @@ +/* + * Copyright 2017 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.transformers; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertArrayEquals; +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.fail; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.Collections; +import java.util.Date; +import java.util.Map; + +import org.hamcrest.Matchers; +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.integration.MessageRejectedException; +import org.springframework.integration.annotation.Transformer; +import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.codec.Codec; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.Transformers; +import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor; +import org.springframework.integration.selector.MetadataStoreSelector; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.handler.annotation.Header; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Artem Bilan + * + * @since 5.0 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class TransformerTests { + + @Autowired + @Qualifier("enricherInput") + private FixedSubscriberChannel enricherInput; + + @Autowired + @Qualifier("enricherInput2") + private FixedSubscriberChannel enricherInput2; + + @Autowired + @Qualifier("enricherInput3") + private FixedSubscriberChannel enricherInput3; + + + @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, instanceOf(TestPojo.class)); + TestPojo result = (TestPojo) payload; + assertEquals("Bar Bar", result.getName()); + assertNotNull(result.getDate()); + assertThat(new Date(), Matchers.greaterThanOrEqualTo(result.getDate())); + } + + @Test + public void testContentEnricher2() { + QueueChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload(new TestPojo("Bar")) + .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) + .build(); + this.enricherInput2.send(message); + Message receive = replyChannel.receive(5000); + assertNotNull(receive); + assertNull(receive.getHeaders().get("foo")); + Object payload = receive.getPayload(); + assertThat(payload, instanceOf(TestPojo.class)); + TestPojo result = (TestPojo) payload; + assertEquals("Bar Bar", result.getName()); + assertNotNull(result.getDate()); + assertThat(new Date(), Matchers.greaterThanOrEqualTo(result.getDate())); + } + + @Test + public void testContentEnricher3() { + QueueChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload(new TestPojo("Bar")) + .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) + .build(); + this.enricherInput3.send(message); + Message receive = replyChannel.receive(5000); + assertNotNull(receive); + assertEquals("Bar Bar", receive.getHeaders().get("foo")); + Object payload = receive.getPayload(); + assertThat(payload, instanceOf(TestPojo.class)); + TestPojo result = (TestPojo) payload; + assertEquals("Bar", result.getName()); + assertNull(result.getDate()); + } + + @Autowired + @Qualifier("encodingFlow.input") + private MessageChannel encodingFlowInput; + + @Autowired + @Qualifier("decodingFlow.input") + private MessageChannel decodingFlowInput; + + @Autowired + @Qualifier("codecReplyChannel") + private PollableChannel codecReplyChannel; + + @Test + public void testCodec() throws Exception { + this.encodingFlowInput.send(new GenericMessage<>("bar")); + Message receive = this.codecReplyChannel.receive(10000); + assertNotNull(receive); + assertThat(receive.getPayload(), instanceOf(byte[].class)); + byte[] transformed = (byte[]) receive.getPayload(); + assertArrayEquals("foo".getBytes(), transformed); + + this.decodingFlowInput.send(new GenericMessage<>(transformed)); + receive = this.codecReplyChannel.receive(10000); + assertNotNull(receive); + assertEquals(42, receive.getPayload()); + } + + + @Autowired + @Qualifier("pojoTransformFlow.input") + private MessageChannel pojoTransformFlowInput; + + @Autowired + private PollableChannel idempotentDiscardChannel; + + @Test + public void transformWithHeader() { + QueueChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("Foo") + .setReplyChannel(replyChannel) + .build(); + this.pojoTransformFlowInput.send(message); + Message receive = replyChannel.receive(10000); + assertNotNull(receive); + assertEquals("FooBar", receive.getPayload()); + + try { + this.pojoTransformFlowInput.send(message); + fail("MessageRejectedException expected"); + } + catch (Exception e) { + assertThat(e, instanceOf(MessageRejectedException.class)); + assertThat(e.getMessage(), containsString("IdempotentReceiver")); + assertThat(e.getMessage(), containsString("rejected duplicate Message")); + } + + assertNotNull(this.idempotentDiscardChannel.receive(10000)); + } + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + public IntegrationFlow enricherFlow() { + return IntegrationFlows.from("enricherInput", true) + .enrich(e -> e.requestChannel("enrichChannel") + .requestPayloadExpression("payload") + .shouldClonePayload(false) + .propertyExpression("name", "payload['name']") + .propertyFunction("date", m -> new Date()) + .headerExpression("foo", "payload['name']") + ) + .get(); + } + + @Bean + public IntegrationFlow enricherFlow2() { + return IntegrationFlows.from("enricherInput2", true) + .enrich(e -> e.requestChannel("enrichChannel") + .requestPayloadExpression("payload") + .shouldClonePayload(false) + .propertyExpression("name", "payload['name']") + .propertyExpression("date", "new java.util.Date()") + ) + .get(); + } + + @Bean + public IntegrationFlow enricherFlow3() { + return IntegrationFlows.from("enricherInput3", true) + .enrich(e -> e.requestChannel("enrichChannel") + .requestPayload(Message::getPayload) + .shouldClonePayload(false) + .>headerFunction("foo", m -> m.getPayload().get("name"))) + .get(); + } + + @Bean + public IntegrationFlow enrichFlow() { + return IntegrationFlows.from("enrichChannel") + .>transform(p -> Collections.singletonMap("name", p.getName() + " Bar")) + .get(); + } + + @Bean + public PollableChannel receivedChannel() { + return new QueueChannel(); + } + + @Bean + public PollableChannel codecReplyChannel() { + return new QueueChannel(); + } + + @Bean + public IntegrationFlow encodingFlow() { + return f -> f + .transform(Transformers.encoding(new MyCodec())) + .channel("codecReplyChannel"); + } + + @Bean + public IntegrationFlow decodingFlow() { + return f -> f + .transform(Transformers.decoding(new MyCodec(), m -> Integer.class)) + .channel("codecReplyChannel"); + } + + @Bean + public IntegrationFlow pojoTransformFlow() { + return f -> f + .enrichHeaders(h -> h.header("Foo", "Bar"), + e -> e.advice(idempotentReceiverInterceptor())) + .transform(new PojoTransformer()); + } + + @Bean + public PollableChannel idempotentDiscardChannel() { + return new QueueChannel(); + } + + @Bean + public IdempotentReceiverInterceptor idempotentReceiverInterceptor() { + IdempotentReceiverInterceptor idempotentReceiverInterceptor = + new IdempotentReceiverInterceptor(new MetadataStoreSelector(m -> m.getPayload().toString())); + idempotentReceiverInterceptor.setDiscardChannel(idempotentDiscardChannel()); + idempotentReceiverInterceptor.setThrowExceptionOnRejection(true); + return idempotentReceiverInterceptor; + } + + + } + + private static final class TestPojo { + + private String name; + + private Date date; + + private TestPojo(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + @SuppressWarnings("unused") + public void setName(String name) { + this.name = name; + } + + public Date getDate() { + return date; + } + + @SuppressWarnings("unused") + public void setDate(Date date) { + this.date = date; + } + + } + + public static class MyCodec implements Codec { + + @Override + public void encode(Object object, OutputStream outputStream) throws IOException { + } + + @Override + public byte[] encode(Object object) throws IOException { + return "foo".getBytes(); + } + + @Override + public T decode(InputStream inputStream, Class type) throws IOException { + return null; + } + + @SuppressWarnings("unchecked") + @Override + public T decode(byte[] bytes, Class type) throws IOException { + return (T) (type.equals(String.class) ? new String(bytes) : + type.equals(Integer.class) ? Integer.valueOf(42) : Integer.valueOf(43)); + } + + } + + public static class PojoTransformer { + + @Transformer + public String transform(String payload, @Header("Foo") String header) { + return payload + header; + } + + } + +}