Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactoring for some DSL methods

Browse Source 
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
This commit is contained in:
Artem Bilan
2017-03-20 17:27:04 -04:00
committed by Gary Russell
parent 70652739b8
commit 3a98b1c974
9 changed files with 454 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-integration-core/src/main/java/org/springframework/integration/dsl/AbstractRouterSpec.java
View File

@@ -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<S extends AbstractRouterSpec<S, R>, R extends AbstractMessageRouter>
extends MessageHandlerSpec<S, R> implements ComponentsRegistration {
protected final List<Object> componentsToRegister = new ArrayList<>();
extends ConsumerEndpointSpec<S, R> {
private boolean defaultToParentFlow;
AbstractRouterSpec(R router) {
this.target = router;
super(router);
}
/**
@@ -52,7 +46,7 @@ public class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, 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<S extends AbstractRouterSpec<S, R>, 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<S extends AbstractRouterSpec<S, R>, 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<S extends AbstractRouterSpec<S, R>, 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<S extends AbstractRouterSpec<S, R>, R extends Ab
return this.defaultToParentFlow;
}
@Override
public Collection<Object> getComponentsToRegister() {
return this.componentsToRegister;
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java
View File

@@ -150,7 +150,7 @@ public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>,
*/
public S transactional(boolean handleMessageAdvice) {
TransactionInterceptor transactionInterceptor = new TransactionInterceptorBuilder(handleMessageAdvice).build();
this.componentToRegister.add(transactionInterceptor);
this.componentsToRegister.add(transactionInterceptor);
return transactional(transactionInterceptor);
}

10
spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java
View File

@@ -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<S extends EndpointSpec<S, F, H>, F extends Be
extends IntegrationComponentSpec<S, Tuple2<F, H>>
implements ComponentsRegistration {
protected final Collection<Object> componentToRegister = new ArrayList<Object>();
protected final Collection<Object> componentsToRegister = new ArrayList<>();
protected H handler;
@@ -89,7 +89,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
public S poller(PollerSpec pollerMetadataSpec) {
Collection<Object> 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<S extends EndpointSpec<S, F, H>, F extends Be
@Override
public Collection<Object> getComponentsToRegister() {
return this.componentToRegister.isEmpty()
return this.componentsToRegister.isEmpty()
? null
: this.componentToRegister;
: this.componentsToRegister;
}
@Override

38
spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java
View File

@@ -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<EnricherSpec, ContentEnricher> {
private final ContentEnricher enricher = new ContentEnricher();
public class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnricher> {
private final Map<String, Expression> propertyExpressions = new HashMap<String, Expression>();
@@ -50,7 +51,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
new HashMap<String, HeaderValueMessageProcessor<?>>();
EnricherSpec() {
super();
super(new ContentEnricher());
}
/**
@@ -59,7 +60,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
* @see ContentEnricher#setRequestChannel(MessageChannel)
*/
public EnricherSpec requestChannel(MessageChannel requestChannel) {
this.enricher.setRequestChannel(requestChannel);
this.handler.setRequestChannel(requestChannel);
return _this();
}
@@ -69,7 +70,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
* @see ContentEnricher#setRequestChannelName(String)
*/
public EnricherSpec requestChannel(String requestChannel) {
this.enricher.setRequestChannelName(requestChannel);
this.handler.setRequestChannelName(requestChannel);
return _this();
}
@@ -79,7 +80,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
* @see ContentEnricher#setReplyChannel(MessageChannel)
*/
public EnricherSpec replyChannel(MessageChannel replyChannel) {
this.enricher.setReplyChannel(replyChannel);
this.handler.setReplyChannel(replyChannel);
return _this();
}
@@ -89,7 +90,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
* @see ContentEnricher#setReplyChannelName(String)
*/
public EnricherSpec replyChannel(String replyChannel) {
this.enricher.setReplyChannelName(replyChannel);
this.handler.setReplyChannelName(replyChannel);
return _this();
}
@@ -99,7 +100,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
* @see ContentEnricher#setRequestTimeout(Long)
*/
public EnricherSpec requestTimeout(Long requestTimeout) {
this.enricher.setRequestTimeout(requestTimeout);
this.handler.setRequestTimeout(requestTimeout);
return _this();
}
@@ -109,7 +110,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
* @see ContentEnricher#setReplyTimeout(Long)
*/
public EnricherSpec replyTimeout(Long replyTimeout) {
this.enricher.setReplyTimeout(replyTimeout);
this.handler.setReplyTimeout(replyTimeout);
return _this();
}
@@ -119,7 +120,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
* @see ContentEnricher#setRequestPayloadExpression(Expression)
*/
public EnricherSpec requestPayloadExpression(String requestPayloadExpression) {
this.enricher.setRequestPayloadExpression(PARSER.parseExpression(requestPayloadExpression));
this.handler.setRequestPayloadExpression(PARSER.parseExpression(requestPayloadExpression));
return _this();
}
@@ -131,7 +132,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
* @see FunctionExpression
*/
public <P> EnricherSpec requestPayload(Function<Message<P>, ?> requestPayloadFunction) {
this.enricher.setRequestPayloadExpression(new FunctionExpression<>(requestPayloadFunction));
this.handler.setRequestPayloadExpression(new FunctionExpression<>(requestPayloadFunction));
return _this();
}
@@ -141,7 +142,7 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
* @see ContentEnricher#setShouldClonePayload(boolean)
*/
public EnricherSpec shouldClonePayload(boolean shouldClonePayload) {
this.enricher.setShouldClonePayload(shouldClonePayload);
this.handler.setShouldClonePayload(shouldClonePayload);
return _this();
}
@@ -279,15 +280,16 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
return _this();
}
@Override
protected ContentEnricher doGet() {
protected Tuple2<ConsumerEndpointFactoryBean, ContentEnricher> 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();
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java
View File

@@ -91,7 +91,7 @@ public final class FilterEndpointSpec extends ConsumerEndpointSpec<FilterEndpoin
DirectChannel channel = new DirectChannel();
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
discardFlow.configure(flowBuilder);
this.componentToRegister.add(flowBuilder.get());
this.componentsToRegister.add(flowBuilder.get());
return discardChannel(channel);
}

277
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -68,7 +68,6 @@ import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.handler.MessageTriggerAction;
import org.springframework.integration.handler.MethodInvokingMessageProcessor;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.router.AbstractMappingMessageRouter;
import org.springframework.integration.router.AbstractMessageRouter;
import org.springframework.integration.router.ExpressionEvaluatingRouter;
import org.springframework.integration.router.MethodInvokingRouter;
@@ -1164,6 +1163,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* .enrich(e -> e.requestChannel("enrichChannel")
* .requestPayload(Message::getPayload)
* .shouldClonePayload(false)
* .autoStartup(false)
* .<Map<String, String>>headerFunction("foo", m -> m.getPayload().get("name")))
* }
* </pre>
@@ -1172,35 +1172,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @see EnricherSpec
*/
public B enrich(Consumer<EnricherSpec> 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:
* <pre class="code">
* {@code
* .enrich(e -> e.requestChannel("enrichChannel")
* .requestPayload(Message::getPayload)
* .shouldClonePayload(false)
* .<Map<String, String>>headerFunction("foo", m -> m.getPayload().get("name")),
* e -> e.autoStartup(false))
* }
* </pre>
* @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<EnricherSpec> enricherConfigurer,
Consumer<GenericEndpointSpec<ContentEnricher>> 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<B extends IntegrationFlowDefinit
*/
public B enrichHeaders(final Map<String, Object> headers,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
return enrichHeaders(new Consumer<HeaderEnricherSpec>() {
@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<B extends IntegrationFlowDefinit
* @return the current {@link IntegrationFlowDefinition}.
*/
public B route(String beanName, String method) {
return this.route(beanName, method, null);
return route(beanName, method, null);
}
/**
@@ -1817,25 +1782,11 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options.
* @return the current {@link IntegrationFlowDefinition}.
*/
public B route(String beanName, String method,
Consumer<RouterSpec<Object, MethodInvokingRouter>> 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<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer,
Consumer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
return this.route(new MethodInvokingRouter(new BeanNameMessageProcessor<Object>(beanName, method)),
routerConfigurer, endpointConfigurer);
public B route(String beanName, String method, Consumer<RouterSpec<Object,
MethodInvokingRouter>> routerConfigurer) {
MethodInvokingRouter methodInvokingRouter =
new MethodInvokingRouter(new BeanNameMessageProcessor<>(beanName, method));
return route(new RouterSpec<>(methodInvokingRouter), routerConfigurer);
}
/**
@@ -1872,23 +1823,6 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public B route(Object service, String methodName,
Consumer<RouterSpec<Object, MethodInvokingRouter>> 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<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer,
Consumer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
MethodInvokingRouter router;
if (StringUtils.hasText(methodName)) {
router = new MethodInvokingRouter(service, methodName);
@@ -1896,7 +1830,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
else {
router = new MethodInvokingRouter(service);
}
return route(router, routerConfigurer, endpointConfigurer);
return route(new RouterSpec<>(router), routerConfigurer);
}
@@ -1907,7 +1841,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @return the current {@link IntegrationFlowDefinition}.
*/
public B route(String expression) {
return this.route(expression, (Consumer<RouterSpec<Object, ExpressionEvaluatingRouter>>) null);
return route(expression, (Consumer<RouterSpec<Object, ExpressionEvaluatingRouter>>) null);
}
/**
@@ -1919,23 +1853,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @return the current {@link IntegrationFlowDefinition}.
*/
public <T> B route(String expression, Consumer<RouterSpec<T, ExpressionEvaluatingRouter>> 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 <T> the target result type.
* @return the current {@link IntegrationFlowDefinition}.
*/
public <T> B route(String expression, Consumer<RouterSpec<T, ExpressionEvaluatingRouter>> routerConfigurer,
Consumer<GenericEndpointSpec<ExpressionEvaluatingRouter>> 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 extends IntegrationFlowDefinit
* @return the current {@link IntegrationFlowDefinition}.
*/
public <S, T> B route(Function<S, T> 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:
* <pre class="code">
* {@code
* .<Integer, Boolean>route(p -> p % 2 == 0,
* m -> m.channelMapping("true", "evenChannel")
* .subFlowMapping("false", f ->
* f.<Integer>handle((p, h) -> p * 3)))
* }
* </pre>
* @param router the {@link Function} to use.
* @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options.
* @param <S> the source payload type.
* @param <T> the target result type.
* @return the current {@link IntegrationFlowDefinition}.
*/
public <S, T> B route(Function<S, T> router, Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
return this.route(null, router, routerConfigurer);
return route(null, router);
}
/**
@@ -1995,32 +1892,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @see LambdaMessageProcessor
*/
public <S, T> B route(Class<S> payloadType, Function<S, T> 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:
* <pre class="code">
* {@code
* .route(Integer.class, p -> p % 2 == 0,
* m -> m.channelMapping("true", "evenChannel")
* .subFlowMapping("false", f ->
* f.<Integer>handle((p, h) -> p * 3)))
* }
* </pre>
* @param payloadType the expected payload type.
* @param router the {@link Function} to use.
* @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options.
* @param <S> the source payload type.
* @param <T> the target result type.
* @return the current {@link IntegrationFlowDefinition}.
* @see LambdaMessageProcessor
*/
public <S, T> B route(Class<S> payloadType, Function<S, T> router,
Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
return this.route(payloadType, router, routerConfigurer, null);
return route(payloadType, router, null);
}
/**
@@ -2033,20 +1905,18 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* .<Integer, Boolean>route(p -> p % 2 == 0,
* m -> m.channelMapping("true", "evenChannel")
* .subFlowMapping("false", f ->
* f.<Integer>handle((p, h) -> p * 3)),
* e -> e.applySequence(false))
* f.<Integer>handle((p, h) -> p * 3))
* .applySequence(false))
* }
* </pre>
* @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 <S> the source payload type.
* @param <T> the target result type.
* @return the current {@link IntegrationFlowDefinition}.
*/
public <S, T> B route(Function<S, T> router, Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer,
Consumer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
return route(null, router, routerConfigurer, endpointConfigurer);
public <S, T> B route(Function<S, T> router, Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
return route(null, router, routerConfigurer);
}
/**
@@ -2059,26 +1929,24 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* .route(Integer.class, p -> p % 2 == 0,
* m -> m.channelMapping("true", "evenChannel")
* .subFlowMapping("false", f ->
* f.<Integer>handle((p, h) -> p * 3)),
* e -> e.applySequence(false))
* f.<Integer>handle((p, h) -> p * 3))
* .applySequence(false))
* }
* </pre>
* @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 <P> the source payload type.
* @param <T> the target result type.
* @return the current {@link IntegrationFlowDefinition}.
* @see LambdaMessageProcessor
*/
public <P, T> B route(Class<P> payloadType, Function<P, T> router,
Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer,
Consumer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
Consumer<RouterSpec<T, MethodInvokingRouter>> 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<B extends IntegrationFlowDefinit
*/
public B route(MessageProcessorSpec<?> messageProcessorSpec,
Consumer<RouterSpec<Object, MethodInvokingRouter>> 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}.
* <pre class="code">
* {@code
* .route(Scripts.script(myScriptResource).lang("groovy").refreshCheckDelay(1000),
* m -> m.channelMapping("true", "evenChannel")
* .subFlowMapping("false", f ->
* f.<Integer>handle((p, h) -> p * 3)),
* e -> e.applySequence(false))
* }
* </pre>
* @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<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer,
Consumer<GenericEndpointSpec<MethodInvokingRouter>> 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 <K> the {@code channelKey mapping} type.
* @param <R> the {@link AbstractMappingMessageRouter} type.
* @return the current {@link IntegrationFlowDefinition}.
*/
public <K, R extends AbstractMappingMessageRouter> B route(R router, Consumer<RouterSpec<K, R>> routerConfigurer,
Consumer<GenericEndpointSpec<R>> endpointConfigurer) {
private <R extends AbstractMessageRouter, S extends AbstractRouterSpec<S, R>> B route(S routerSpec,
Consumer<S> routerConfigurer) {
RouterSpec<K, R> routerSpec = new RouterSpec<>(router);
if (routerConfigurer != null) {
routerConfigurer.accept(routerSpec);
}
return route(router, routerSpec, endpointConfigurer);
}
private <R extends AbstractMessageRouter, S extends AbstractRouterSpec<S, R>> B route(R router,
S routerSpec, Consumer<GenericEndpointSpec<R>> endpointConfigurer) {
route(router, endpointConfigurer);
final BridgeHandler bridgeHandler = new BridgeHandler();
BridgeHandler bridgeHandler = new BridgeHandler();
boolean registerSubflowBridge = false;
Collection<Object> componentsToRegister = routerSpec.getComponentsToRegister();
if (!CollectionUtils.isEmpty(componentsToRegister)) {
@@ -2188,6 +2013,13 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
}
}
if (componentsToRegister != null) {
componentsToRegister.clear();
}
register(routerSpec, null);
if (routerSpec.isDefaultToParentFlow()) {
routerSpec.defaultOutputChannel(new FixedSubscriberChannel(bridgeHandler));
registerSubflowBridge = true;
@@ -2217,37 +2049,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @return the current {@link IntegrationFlowDefinition}.
*/
public B routeToRecipients(Consumer<RecipientListRouterSpec> 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:
* <pre class="code">
* {@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))
* }
* </pre>
* @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<RecipientListRouterSpec> routerConfigurer,
Consumer<GenericEndpointSpec<RecipientListRouter>> 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<B extends IntegrationFlowDefinit
gatherer.accept(aggregatorSpec);
}
RecipientListRouter recipientListRouter = recipientListRouterSpec.get();
addComponent(recipientListRouterSpec);
RecipientListRouter recipientListRouter = recipientListRouterSpec.get().getT2();
addComponent(recipientListRouter)
.addComponents(recipientListRouterSpec.getComponentsToRegister());
AggregatingMessageHandler aggregatingMessageHandler = aggregatorSpec.get().getT2();
addComponent(aggregatingMessageHandler);
ScatterGatherHandler messageHandler = new ScatterGatherHandler(recipientListRouter, aggregatingMessageHandler);

12
spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.java
View File

@@ -62,7 +62,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
return recipient(channelName, PARSER.parseExpression(expression));
}
else {
this.target.addRecipient(channelName);
this.handler.addRecipient(channelName);
return _this();
}
}
@@ -75,7 +75,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
*/
public RecipientListRouterSpec recipient(String channelName, Expression expression) {
ExpressionEvaluatingSelector selector = new ExpressionEvaluatingSelector(expression);
this.target.addRecipient(channelName, selector);
this.handler.addRecipient(channelName, selector);
this.componentsToRegister.add(selector);
return _this();
}
@@ -108,7 +108,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
? new MethodInvokingSelector(new LambdaMessageProcessor(selector, null))
: new MethodInvokingSelector(selector);
}
this.target.addRecipient(channelName, messageSelector);
this.handler.addRecipient(channelName, messageSelector);
return _this();
}
@@ -145,11 +145,11 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
public RecipientListRouterSpec recipient(MessageChannel channel, Expression expression) {
if (expression != null) {
ExpressionEvaluatingSelector selector = new ExpressionEvaluatingSelector(expression);
this.target.addRecipient(channel, selector);
this.handler.addRecipient(channel, selector);
this.componentsToRegister.add(selector);
}
else {
this.target.addRecipient(channel);
this.handler.addRecipient(channel);
}
return _this();
}
@@ -182,7 +182,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
? new MethodInvokingSelector(new LambdaMessageProcessor(selector, null))
: new MethodInvokingSelector(selector);
}
this.target.addRecipient(channel, messageSelector);
this.handler.addRecipient(channel, messageSelector);
return _this();
}

27
spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java
View File

@@ -51,9 +51,11 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
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<K, R extends AbstractMappingMessageRouter>
* @see AbstractMappingMessageRouter#setResolutionRequired(boolean)
*/
public RouterSpec<K, R> resolutionRequired(boolean resolutionRequired) {
this.target.setResolutionRequired(resolutionRequired);
this.handler.setResolutionRequired(resolutionRequired);
return _this();
}
@@ -73,9 +75,10 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
* @see AbstractMappingMessageRouter#setPrefix(String)
*/
public RouterSpec<K, R> 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<K, R extends AbstractMappingMessageRouter>
* @see AbstractMappingMessageRouter#setSuffix(String)
*/
public RouterSpec<K, R> 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<K, R extends AbstractMappingMessageRouter>
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<K, R extends AbstractMappingMessageRouter>
@Override
public Collection<Object> 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();
}

357
spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java Normal file
View File

@@ -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)
.<Map<String, String>>headerFunction("foo", m -> m.getPayload().get("name")))
.get();
}
@Bean
public IntegrationFlow enrichFlow() {
return IntegrationFlows.from("enrichChannel")
.<TestPojo, Map<?, ?>>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> T decode(InputStream inputStream, Class<T> type) throws IOException {
return null;
}
@SuppressWarnings("unchecked")
@Override
public <T> T decode(byte[] bytes, Class<T> 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;
}
}
}