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

DSL: RecipientListRouter: Add SubFlow support

Browse Source
This commit is contained in:
Artem Bilan
2014-10-31 13:43:26 +02:00
parent be9767b8a9
commit e6af7f7eba
4 changed files with 110 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslRecipientListRouter.java
View File

@@ -39,6 +39,11 @@ class DslRecipientListRouter extends RecipientListRouter {
private final Map<String, MessageSelector> selectorRecipientMap = new HashMap<String, MessageSelector>();
private final Map<MessageChannel, String> channelExpressionRecipientMap = new HashMap<MessageChannel, String>();
private final Map<MessageChannel, MessageSelector> channelSelectorRecipientMap =
new HashMap<MessageChannel, MessageSelector>();
void add(String channelName, String expression) {
this.expressionRecipientMap.put(channelName, expression);
}
@@ -47,12 +52,12 @@ class DslRecipientListRouter extends RecipientListRouter {
this.selectorRecipientMap.put(channelName, selector);
}
Map<String, Object> get() {
Map<String, Object> recipients =
new HashMap<String, Object>(this.expressionRecipientMap.size() + this.selectorRecipientMap.size());
recipients.putAll(this.expressionRecipientMap);
recipients.putAll(this.selectorRecipientMap);
return recipients;
void add(MessageChannel channel, String expression) {
this.channelExpressionRecipientMap.put(channel, expression);
}
void add(MessageChannel channel, MessageSelector selector) {
this.channelSelectorRecipientMap.put(channel, selector);
}
@Override
@@ -67,13 +72,28 @@ class DslRecipientListRouter extends RecipientListRouter {
this.selectorRecipientMap.put(recipient.getKey(), selector);
}
List<Recipient> recipients = new ArrayList<Recipient>(this.selectorRecipientMap.size());
for (Map.Entry<MessageChannel, String> recipient : this.channelExpressionRecipientMap.entrySet()) {
ExpressionEvaluatingSelector selector = null;
String expression = recipient.getValue();
if (StringUtils.hasText(expression)) {
selector = new ExpressionEvaluatingSelector(expression);
selector.setBeanFactory(this.getBeanFactory());
}
this.channelSelectorRecipientMap.put(recipient.getKey(), selector);
}
List<Recipient> recipients = new ArrayList<Recipient>(this.selectorRecipientMap.size()
+ this.channelSelectorRecipientMap.size());
for (Map.Entry<String, MessageSelector> entry : selectorRecipientMap.entrySet()) {
recipients.add(new DslRecipient(new MessageChannelReference(entry.getKey()), entry.getValue()));
}
this.setRecipients(recipients);
for (Map.Entry<MessageChannel, MessageSelector> entry : channelSelectorRecipientMap.entrySet()) {
recipients.add(new Recipient(entry.getKey(), entry.getValue()));
}
setRecipients(recipients);
super.onInit();
}

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

@@ -630,7 +630,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
public B routeToRecipients(Consumer<RecipientListRouterSpec> routerConfigurer) {
return this.routeToRecipients(routerConfigurer, null);
return routeToRecipients(routerConfigurer, null);
}
public B routeToRecipients(Consumer<RecipientListRouterSpec> routerConfigurer,
@@ -638,18 +638,17 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
Assert.notNull(routerConfigurer);
RecipientListRouterSpec spec = new RecipientListRouterSpec();
routerConfigurer.accept(spec);
DslRecipientListRouter recipientListRouter = (DslRecipientListRouter) spec.get();
Assert.notEmpty(recipientListRouter.get(), "recipient list must not be empty");
return this.route(recipientListRouter, endpointConfigurer);
addComponents(spec.getComponentsToRegister());
return route(spec.get(), endpointConfigurer);
}
public B route(AbstractMessageRouter router) {
return this.route(router, null);
return route(router, null);
}
public <R extends AbstractMessageRouter> B route(R router,
Consumer<GenericEndpointSpec<R>> endpointConfigurer) {
return this.handle(router, endpointConfigurer);
return handle(router, endpointConfigurer);
}
public B gateway(String requestChannel) {

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

@@ -16,14 +16,23 @@
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.core.MessageSelector;
import org.springframework.integration.dsl.core.ComponentsRegistration;
import org.springframework.integration.router.RecipientListRouter;
import org.springframework.util.Assert;
/**
* @author Artem Bilan
*/
public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRouterSpec, RecipientListRouter> {
public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRouterSpec, RecipientListRouter>
implements ComponentsRegistration {
private final List<Object> subFlows = new ArrayList<Object>();
RecipientListRouterSpec() {
super(new DslRecipientListRouter());
@@ -41,4 +50,31 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
return _this();
}
public RecipientListRouterSpec recipientFlow(MessageSelector selector, IntegrationFlow subFlow) {
Assert.notNull(subFlow);
DirectChannel channel = populateSubFlow(subFlow);
((DslRecipientListRouter) this.target).add(channel, selector);
return _this();
}
public RecipientListRouterSpec recipientFlow(String expression, IntegrationFlow subFlow) {
Assert.notNull(subFlow);
DirectChannel channel = populateSubFlow(subFlow);
((DslRecipientListRouter) this.target).add(channel, expression);
return _this();
}
private DirectChannel populateSubFlow(IntegrationFlow subFlow) {
DirectChannel channel = new DirectChannel();
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
subFlow.accept(flowBuilder);
this.subFlows.add(flowBuilder.get());
return channel;
}
@Override
public Collection<Object> getComponentsToRegister() {
return subFlows;
}
}

46
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java
View File

@@ -231,6 +231,14 @@ public class IntegrationFlowTests {
@Qualifier("recipientListInput")
private MessageChannel recipientListInput;
@Autowired
@Qualifier("recipientListSubFlow1Result")
private PollableChannel recipientListSubFlow1Result;
@Autowired
@Qualifier("recipientListSubFlow2Result")
private PollableChannel recipientListSubFlow2Result;
@Autowired
@Qualifier("defaultOutputChannel")
private PollableChannel defaultOutputChannel;
@@ -687,6 +695,7 @@ public class IntegrationFlowTests {
Message<String> fooMessage = MessageBuilder.withPayload("fooPayload").setHeader("recipient", true).build();
Message<String> barMessage = MessageBuilder.withPayload("barPayload").setHeader("recipient", true).build();
Message<String> bazMessage = new GenericMessage<>("baz");
Message<String> badMessage = new GenericMessage<>("badPayload");
this.recipientListInput.send(fooMessage);
@@ -696,21 +705,39 @@ public class IntegrationFlowTests {
Message<?> result1b = this.barChannel.receive(2000);
assertNotNull(result1b);
assertEquals("foo", result1b.getPayload());
Message<?> result1c = this.recipientListSubFlow1Result.receive(2000);
assertNotNull(result1c);
assertEquals("FOO", result1c.getPayload());
assertNull(this.recipientListSubFlow2Result.receive(0));
this.recipientListInput.send(barMessage);
assertNull(this.fooChannel.receive(0));
assertNull(this.recipientListSubFlow2Result.receive(0));
Message<?> result2b = this.barChannel.receive(2000);
assertNotNull(result2b);
assertEquals("bar", result2b.getPayload());
Message<?> result2c = this.recipientListSubFlow1Result.receive(2000);
assertNotNull(result1c);
assertEquals("BAR", result2c.getPayload());
this.recipientListInput.send(bazMessage);
assertNull(this.fooChannel.receive(0));
assertNull(this.barChannel.receive(0));
Message<?> result3c = this.recipientListSubFlow1Result.receive(2000);
assertNotNull(result3c);
assertEquals("BAZ", result3c.getPayload());
Message<?> result4c = this.recipientListSubFlow2Result.receive(2000);
assertNotNull(result4c);
assertEquals("Hello baz", result4c.getPayload());
this.recipientListInput.send(badMessage);
assertNull(this.fooChannel.receive(0));
assertNull(this.barChannel.receive(0));
Message<?> result3c = this.defaultOutputChannel.receive(2000);
assertNotNull(result3c);
assertEquals("bad", result3c.getPayload());
assertNull(this.recipientListSubFlow1Result.receive(0));
assertNull(this.recipientListSubFlow2Result.receive(0));
Message<?> resultD = this.defaultOutputChannel.receive(2000);
assertNotNull(resultD);
assertEquals("bad", resultD.getPayload());
}
@Test
@@ -990,10 +1017,17 @@ public class IntegrationFlowTests {
public IntegrationFlow recipientListFlow() {
return IntegrationFlows.from("recipientListInput")
.<String, String>transform(p -> p.replaceFirst("Payload", ""))
.routeToRecipients(r -> r.recipient("foo-channel", "'foo' == payload")
.routeToRecipients(r -> r
.recipient("foo-channel", "'foo' == payload")
.recipient("bar-channel", m ->
m.getHeaders().containsKey("recipient")
&& (boolean) m.getHeaders().get("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")))
.recipientFlow(m -> "baz".equals(m.getPayload()),
f -> f.transform("Hello "::concat)
.channel(c -> c.queue("recipientListSubFlow2Result"))))
.channel("defaultOutputChannel")
.get();
}