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

INT-578 adding namespace support for dynamic recipient list routers (using an expression)

Browse Source
This commit is contained in:
Mark Fisher
2010-07-03 22:05:45 +00:00
parent d26c9eb197
commit 1553df8a54
4 changed files with 89 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-integration-core/src/main/java/org/springframework/integration/config/xml/RecipientListRouterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 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.
@@ -20,17 +20,20 @@ import java.util.List;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
* Parser for the <recipient-list-router/> element.
*
* @author Oleg Zhurakousky
* @author Mark Fisher
* @since 1.0.3
*/
public class RecipientListRouterParser extends AbstractConsumerEndpointParser {
@@ -43,11 +46,22 @@ public class RecipientListRouterParser extends AbstractConsumerEndpointParser {
List<Element> childElements = DomUtils.getChildElementsByTagName(element, "recipient");
Assert.notEmpty(childElements,
"At least one recipient channel must be defined (e.g., <recipient channel=\"channel1\"/>).");
ManagedList channelList = new ManagedList();
ManagedList recipientList = new ManagedList();
for (Element childElement : childElements) {
channelList.add(new RuntimeBeanReference(childElement.getAttribute("channel")));
BeanDefinitionBuilder recipientBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.router.RecipientListRouter.Recipient");
recipientBuilder.addConstructorArgReference(childElement.getAttribute("channel"));
String expression = childElement.getAttribute("expression");
if (StringUtils.hasText(expression)) {
BeanDefinition selectorDef = new RootBeanDefinition(
"org.springframework.integration.filter.ExpressionEvaluatingSelector");
selectorDef.getConstructorArgumentValues().addGenericArgumentValue(expression);
String selectorBeanName = parserContext.getReaderContext().registerWithGeneratedName(selectorDef);
recipientBuilder.addConstructorArgReference(selectorBeanName);
}
recipientList.add(recipientBuilder.getBeanDefinition());
}
recipientListRouterBuilder.addPropertyValue("channels", channelList);
recipientListRouterBuilder.addPropertyValue("recipients", recipientList);
IntegrationNamespaceUtils.setValueIfAttributeDefined(recipientListRouterBuilder, element, "timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(recipientListRouterBuilder, element, "ignore-send-failures");
IntegrationNamespaceUtils.setValueIfAttributeDefined(recipientListRouterBuilder, element, "apply-sequence");

73
spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java
View File

@@ -18,8 +18,6 @@ package org.springframework.integration.router;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -59,32 +57,29 @@ import org.springframework.util.Assert;
*/
public class RecipientListRouter extends AbstractMessageRouter implements InitializingBean {
private volatile Map<MessageSelector, ? extends Collection<? extends MessageChannel>> channelMap;
private volatile List<Recipient> recipients;
/**
* Set the output channels of this router.
*
* @param channels
* Set the channels for this router. Either call this method or
* {@link #setRecipients(List)} but not both. If MessageSelectors
* should be considered, then use {@link #setRecipients(List)}.
*/
public void setChannels(List<MessageChannel> channels) {
Assert.notEmpty(channels, "channels must not be empty");
MessageSelector selector = new MessageSelector() {
public boolean accept(Message<?> message) {
return true;
}
};
this.setChannelMap(Collections.singletonMap(selector, channels));
List<Recipient> recipients = new ArrayList<Recipient>();
for (MessageChannel channel : channels) {
recipients.add(new Recipient(channel));
}
this.setRecipients(recipients);
}
/**
* Set a custom map of selectors to channels. This allows a configuration where groups of channels are used for
* messages matching a particular filter.
*
* @param channelMap
* Set the recipients for this router.
*/
public void setChannelMap(Map<MessageSelector, ? extends Collection<? extends MessageChannel>> channelMap) {
this.channelMap = channelMap;
public void setRecipients(List<Recipient> recipients) {
Assert.notEmpty(recipients, "recipients must not be empty");
this.recipients = recipients;
}
@Override
@@ -94,20 +89,44 @@ public class RecipientListRouter extends AbstractMessageRouter implements Initia
@Override
public final void onInit() {
Assert.notEmpty(this.channelMap, "a non-empty channel map is required");
Assert.notEmpty(this.recipients, "a non-empty recipient list is required");
}
@Override
protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
List<MessageChannel> recipients = new ArrayList<MessageChannel>();
Map<MessageSelector, Collection<? extends MessageChannel>> map =
new HashMap<MessageSelector, Collection<? extends MessageChannel>>(this.channelMap);
for (MessageSelector selector : map.keySet()) {
if (selector.accept(message)) {
recipients.addAll(map.get(selector));
List<MessageChannel> channels = new ArrayList<MessageChannel>();
List<Recipient> recipientList = this.recipients;
for (Recipient recipient : recipientList) {
if (recipient.accept(message)) {
channels.add(recipient.getChannel());
}
}
return recipients;
return channels;
}
public static class Recipient {
private final MessageChannel channel;
private final MessageSelector selector;
public Recipient(MessageChannel channel) {
this(channel, null);
}
public Recipient(MessageChannel channel, MessageSelector selector) {
this.channel = channel;
this.selector = selector;
}
public boolean accept(Message<?> message) {
return (this.selector != null ? this.selector.accept(message) : true);
}
public MessageChannel getChannel() {
return this.channel;
}
}
}

11
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -1642,7 +1642,16 @@
<xsd:sequence>
<xsd:element name="recipient" minOccurs="1" maxOccurs="unbounded">
<xsd:complexType>
<xsd:attribute name="channel" type="xsd:string">
<xsd:attribute name="expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
An expression to be evaluated to determine if this recipient should be included in the recipient
list for a given input Message. The evaluation result of the expression must be a boolean.
If this attribute is not defined, the channel will always be among the list of recipients.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">

42
spring-integration-core/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java
View File

@@ -19,14 +19,10 @@ package org.springframework.integration.router;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.Test;
@@ -36,6 +32,7 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.router.RecipientListRouter.Recipient;
import org.springframework.integration.selector.MessageSelector;
/**
@@ -54,13 +51,11 @@ public class RecipientListRouterTests {
RecipientListRouter router = new RecipientListRouter();
router.setChannels(channels);
router.afterPropertiesSet();
Map<MessageSelector, Collection<MessageChannel>> channelMap =
(Map<MessageSelector, Collection<MessageChannel>>)
new DirectFieldAccessor(router).getPropertyValue("channelMap");
Collection<MessageChannel> channelList = channelMap.values().iterator().next();
assertEquals(2, channelList.size());
assertTrue(channelList.contains(channel1));
assertTrue(channelList.contains(channel2));
List<Recipient> recipients = (List<Recipient>)
new DirectFieldAccessor(router).getPropertyValue("recipients");
assertEquals(2, recipients.size());
assertEquals(channel1, new DirectFieldAccessor(recipients.get(0)).getPropertyValue("channel"));
assertEquals(channel2, new DirectFieldAccessor(recipients.get(1)).getPropertyValue("channel"));
}
@Test
@@ -359,27 +354,20 @@ public class RecipientListRouterTests {
}
@Test
public void selectors() {
public void recipientsWithSelectors() {
QueueChannel channel1 = new QueueChannel();
QueueChannel channel2 = new QueueChannel();
QueueChannel channel3 = new QueueChannel();
QueueChannel channel4 = new QueueChannel();
QueueChannel channel5 = new QueueChannel();
QueueChannel channel6 = new QueueChannel();
List<Recipient> recipients = new ArrayList<Recipient>();
recipients.add(new Recipient(channel1, new AlwaysTrueSelector()));
recipients.add(new Recipient(channel2, new AlwaysFalseSelector()));
recipients.add(new Recipient(channel3));
recipients.add(new Recipient(channel4));
recipients.add(new Recipient(channel5, new AlwaysFalseSelector()));
RecipientListRouter router = new RecipientListRouter();
Map<MessageSelector, List<? extends MessageChannel>> channelMap =
new HashMap<MessageSelector, List<? extends MessageChannel>>();
channelMap.put(new AlwaysTrueSelector(), Collections.singletonList(channel1));
channelMap.put(new AlwaysFalseSelector(), Collections.singletonList(channel2));
List<QueueChannel> acceptList = new ArrayList<QueueChannel>();
acceptList.add(channel3);
acceptList.add(channel4);
channelMap.put(new AlwaysTrueSelector(), acceptList);
List<QueueChannel> rejectList = new ArrayList<QueueChannel>();
rejectList.add(channel5);
rejectList.add(channel6);
channelMap.put(new AlwaysFalseSelector(), rejectList);
router.setChannelMap(channelMap);
router.setRecipients(recipients);
Message<?> message = new StringMessage("test");
router.handleMessage(message);
Message<?> reply1 = channel1.receive(0);
@@ -392,8 +380,6 @@ public class RecipientListRouterTests {
assertEquals(message, reply4);
Message<?> reply5 = channel5.receive(0);
assertNull(reply5);
Message<?> reply6 = channel6.receive(0);
assertNull(reply6);
}