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

Added namespace support for <priority-channel/> (INT-131).

Browse Source
This commit is contained in:
Mark Fisher
2008-03-04 22:24:55 +00:00
parent 0d8dc5ba81
commit 1bed4b19e9
6 changed files with 97 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java
View File

@@ -29,6 +29,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.PriorityChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.channel.interceptor.MessageSelectingInterceptor;
import org.springframework.integration.message.selector.MessageSelector;
@@ -56,9 +57,13 @@ public class ChannelParser implements BeanDefinitionParser {
private static final String INTERCEPTORS_PROPERTY = "interceptors";
private static final String COMPARATOR_REF_ATTRIBUTE = "comparator-ref";
public BeanDefinition parse(Element element, ParserContext parserContext) {
RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class);
boolean isPriorityChannel = (element.getLocalName().equals("priority-channel"));
Class<?> channelClass = (isPriorityChannel) ? PriorityChannel.class : SimpleChannel.class;
RootBeanDefinition channelDef = new RootBeanDefinition(channelClass);
channelDef.setSource(parserContext.extractSource(element));
boolean isPublishSubscribe = "true".equals(element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE));
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(isPublishSubscribe);
@@ -81,6 +86,12 @@ public class ChannelParser implements BeanDefinitionParser {
int capacity = (StringUtils.hasText(capAttr)) ? Integer.parseInt(capAttr) : SimpleChannel.DEFAULT_CAPACITY;
channelDef.getConstructorArgumentValues().addIndexedArgumentValue(0, capacity);
channelDef.getConstructorArgumentValues().addIndexedArgumentValue(1, dispatcherPolicy);
if (isPriorityChannel) {
String comparatorRef = element.getAttribute(COMPARATOR_REF_ATTRIBUTE);
if (StringUtils.hasText(comparatorRef)) {
channelDef.getConstructorArgumentValues().addIndexedArgumentValue(2, new RuntimeBeanReference(comparatorRef));
}
}
String datatypeAttr = element.getAttribute(DATATYPE_ATTRIBUTE);
if (StringUtils.hasText(datatypeAttr)) {
String[] datatypes = StringUtils.commaDelimitedListToStringArray(datatypeAttr);

1
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java
View File

@@ -48,6 +48,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
registerBeanDefinitionParser("message-bus", new MessageBusParser());
registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenParser());
registerBeanDefinitionParser("channel", new ChannelParser());
registerBeanDefinitionParser("priority-channel", new ChannelParser());
registerBeanDefinitionParser("source-adapter", new ChannelAdapterParser(true));
registerBeanDefinitionParser("target-adapter", new ChannelAdapterParser(false));
registerBeanDefinitionParser("endpoint", new EndpointParser());

35
spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -44,24 +44,39 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="channel">
<xsd:element name="channel" type="channelType"/>
<xsd:element name="priority-channel">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a message channel.
Defines a channel with priority-ordering for message reception.
</xsd:documentation>
</xsd:annotation>
<xsd:sequence>
<xsd:element ref="dispatcher-policy" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="interceptor" minOccurs="0" maxOccurs="unbounded"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" use="required"/>
<xsd:attribute name="capacity" type="xsd:integer"/>
<xsd:attribute name="publish-subscribe" type="xsd:boolean" default="false"/>
<xsd:attribute name="datatype" type="xsd:string"/>
<xsd:complexContent>
<xsd:extension base="channelType">
<xsd:attribute name="comparator-ref" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="channelType">
<xsd:annotation>
<xsd:documentation>
Defines a message channel.
</xsd:documentation>
</xsd:annotation>
<xsd:sequence>
<xsd:element ref="dispatcher-policy" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="interceptor" minOccurs="0" maxOccurs="unbounded"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" use="required"/>
<xsd:attribute name="capacity" type="xsd:integer"/>
<xsd:attribute name="publish-subscribe" type="xsd:boolean" default="false"/>
<xsd:attribute name="datatype" type="xsd:string"/>
</xsd:complexType>
<xsd:element name="interceptor">
<xsd:complexType>
<xsd:annotation>

2
spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java
View File

@@ -91,7 +91,7 @@ public class PriorityChannelTests {
}
private static class StringPayloadComparator implements Comparator<Message<?>> {
public static class StringPayloadComparator implements Comparator<Message<?>> {
public int compare(Message<?> message1, Message<?> message2) {
String s1 = (String) message1.getPayload();

41
spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java
View File

@@ -205,6 +205,47 @@ public class ChannelParserTests {
assertEquals(1, interceptor.getReceiveCount());
}
@Test
public void testPriorityChannelWithDefaultComparator() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"priorityChannelParserTests.xml", this.getClass());
MessageChannel channel = (MessageChannel) context.getBean("priorityChannelWithDefaultComparator");
Message<?> lowPriorityMessage = new StringMessage("low");
lowPriorityMessage.getHeader().setPriority(777);
Message<?> midPriorityMessage = new StringMessage("mid");
midPriorityMessage.getHeader().setPriority(77);
Message<?> highPriorityMessage = new StringMessage("high");
highPriorityMessage.getHeader().setPriority(7);
channel.send(lowPriorityMessage);
channel.send(highPriorityMessage);
channel.send(midPriorityMessage);
Message<?> reply1 = channel.receive(0);
Message<?> reply2 = channel.receive(0);
Message<?> reply3 = channel.receive(0);
assertEquals("high", reply1.getPayload());
assertEquals("mid", reply2.getPayload());
assertEquals("low", reply3.getPayload());
}
@Test
public void testPriorityChannelWithCustomComparator() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"priorityChannelParserTests.xml", this.getClass());
MessageChannel channel = (MessageChannel) context.getBean("priorityChannelWithCustomComparator");
channel.send(new StringMessage("C"));
channel.send(new StringMessage("A"));
channel.send(new StringMessage("D"));
channel.send(new StringMessage("B"));
Message<?> reply1 = channel.receive(0);
Message<?> reply2 = channel.receive(0);
Message<?> reply3 = channel.receive(0);
Message<?> reply4 = channel.receive(0);
assertEquals("A", reply1.getPayload());
assertEquals("B", reply2.getPayload());
assertEquals("C", reply3.getPayload());
assertEquals("D", reply4.getPayload());
}
private static class TestHandler implements MessageHandler {

17
spring-integration-core/src/test/java/org/springframework/integration/config/priorityChannelParserTests.xml Normal file
View File

@@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<priority-channel id="priorityChannelWithDefaultComparator"/>
<priority-channel id="priorityChannelWithCustomComparator" comparator-ref="stringPayloadComparator"/>
<beans:bean id="stringPayloadComparator"
class="org.springframework.integration.channel.PriorityChannelTests$StringPayloadComparator"/>
</beans:beans>