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

Added namespace support for the <default-concurrency/> sub-element of <message-bus/> (INT-106).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-25 16:14:24 +00:00
parent 3fc03d7795
commit 7cb5c918c5
7 changed files with 155 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -78,7 +78,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
private volatile ScheduledExecutorService executor;
private volatile ConcurrencyPolicy defaultConcurrencyPolicy = new ConcurrencyPolicy(1, 10);
private volatile ConcurrencyPolicy defaultConcurrencyPolicy;
private volatile boolean autoCreateChannels;
@@ -175,6 +175,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
if (this.getErrorChannel() == null) {
this.setErrorChannel(new DefaultErrorChannel());
}
if (this.defaultConcurrencyPolicy == null) {
this.defaultConcurrencyPolicy = new ConcurrencyPolicy();
}
if (this.executor == null) {
this.executor = new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE);
}
@@ -241,9 +244,15 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
}
public void registerEndpoint(String name, MessageEndpoint endpoint) {
if (!this.initialized) {
this.initialize();
}
if (endpoint instanceof ChannelRegistryAware) {
((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry);
}
if (endpoint.getConcurrencyPolicy() == null && endpoint instanceof DefaultMessageEndpoint) {
((DefaultMessageEndpoint) endpoint).setConcurrencyPolicy(this.defaultConcurrencyPolicy);
}
this.endpoints.put(name, endpoint);
if (this.isRunning()) {
activateEndpoint(endpoint);

26
spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java
View File

@@ -88,14 +88,6 @@ public class EndpointParser implements BeanDefinitionParser {
private static final String CONCURRENCY_ELEMENT = "concurrency";
private static final String CORE_SIZE_ATTRIBUTE = "core";
private static final String MAX_SIZE_ATTRIBUTE = "max";
private static final String QUEUE_CAPACITY_ATTRIBUTE = "queue-capacity";
private static final String KEEP_ALIVE_ATTRIBUTE = "keep-alive";
private static final String CONCURRENCY_POLICY_PROPERTY = "concurrencyPolicy";
@@ -187,23 +179,7 @@ public class EndpointParser implements BeanDefinitionParser {
}
private void parseConcurrencyPolicy(Element concurrencyElement, RootBeanDefinition endpointDef) {
String coreSize = concurrencyElement.getAttribute(CORE_SIZE_ATTRIBUTE);
String maxSize = concurrencyElement.getAttribute(MAX_SIZE_ATTRIBUTE);
String queueCapacity = concurrencyElement.getAttribute(QUEUE_CAPACITY_ATTRIBUTE);
String keepAlive = concurrencyElement.getAttribute(KEEP_ALIVE_ATTRIBUTE);
ConcurrencyPolicy policy = new ConcurrencyPolicy();
if (StringUtils.hasText(coreSize)) {
policy.setCoreSize(Integer.parseInt(coreSize));
}
if (StringUtils.hasText(maxSize)) {
policy.setMaxSize(Integer.parseInt(maxSize));
}
if (StringUtils.hasText(queueCapacity)) {
policy.setQueueCapacity(Integer.parseInt(queueCapacity));
}
if (StringUtils.hasText(keepAlive)) {
policy.setKeepAliveSeconds(Integer.parseInt(keepAlive));
}
ConcurrencyPolicy policy = IntegrationNamespaceUtils.parseConcurrencyPolicy(concurrencyElement);
endpointDef.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy);
}

61
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* Copyright 2002-2007 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.config;
import org.w3c.dom.Element;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.util.StringUtils;
/**
* Shared utility methods for integration namespace parsers.
*
* @author Mark Fisher
*/
public abstract class IntegrationNamespaceUtils {
private static final String CORE_SIZE_ATTRIBUTE = "core";
private static final String MAX_SIZE_ATTRIBUTE = "max";
private static final String QUEUE_CAPACITY_ATTRIBUTE = "queue-capacity";
private static final String KEEP_ALIVE_ATTRIBUTE = "keep-alive";
public static ConcurrencyPolicy parseConcurrencyPolicy(Element element) {
ConcurrencyPolicy policy = new ConcurrencyPolicy();
String coreSize = element.getAttribute(CORE_SIZE_ATTRIBUTE);
String maxSize = element.getAttribute(MAX_SIZE_ATTRIBUTE);
String queueCapacity = element.getAttribute(QUEUE_CAPACITY_ATTRIBUTE);
String keepAlive = element.getAttribute(KEEP_ALIVE_ATTRIBUTE);
if (StringUtils.hasText(coreSize)) {
policy.setCoreSize(Integer.parseInt(coreSize));
}
if (StringUtils.hasText(maxSize)) {
policy.setMaxSize(Integer.parseInt(maxSize));
}
if (StringUtils.hasText(queueCapacity)) {
policy.setQueueCapacity(Integer.parseInt(queueCapacity));
}
if (StringUtils.hasText(keepAlive)) {
policy.setKeepAliveSeconds(Integer.parseInt(keepAlive));
}
return policy;
}
}

22
spring-integration-core/src/main/java/org/springframework/integration/config/MessageBusParser.java
View File

@@ -17,6 +17,8 @@
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
@@ -26,6 +28,7 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.util.StringUtils;
/**
@@ -41,6 +44,10 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
private static final String ERROR_CHANNEL_ATTRIBUTE = "error-channel";
private static final String DEFAULT_CONCURRENCY_ELEMENT = "default-concurrency";
private static final String DEFAULT_CONCURRENCY_PROPERTY = "defaultConcurrencyPolicy";
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
@@ -69,6 +76,21 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
beanDefinition.addPropertyReference(Conventions.attributeNameToPropertyName(
ERROR_CHANNEL_ATTRIBUTE), errorChannelRef);
}
this.registerDefaultConcurrencyIfAvailable(beanDefinition, element);
}
private void registerDefaultConcurrencyIfAvailable(BeanDefinitionBuilder beanDefinition, Element element) {
NodeList childNodes = element.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
Node child = childNodes.item(i);
if (child.getNodeType() == Node.ELEMENT_NODE) {
String localName = child.getLocalName();
if (DEFAULT_CONCURRENCY_ELEMENT.equals(localName)) {
ConcurrencyPolicy policy = IntegrationNamespaceUtils.parseConcurrencyPolicy((Element) child);
beanDefinition.addPropertyValue(DEFAULT_CONCURRENCY_PROPERTY, policy);
}
}
}
}
}

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

@@ -24,6 +24,9 @@
Defines a message bus.
</xsd:documentation>
</xsd:annotation>
<xsd:sequence>
<xsd:element name="default-concurrency" type="concurrencyType" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="auto-startup" type="xsd:boolean"/>
<xsd:attribute name="auto-create-channels" type="xsd:boolean"/>
<xsd:attribute name="error-channel" type="xsd:string"/>
@@ -116,7 +119,7 @@
<xsd:extension base="beans:identifiedType">
<xsd:sequence>
<xsd:element ref="schedule" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="concurrency" minOccurs="0" maxOccurs="1"/>
<xsd:element name="concurrency" type="concurrencyType" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="selector" minOccurs="0" maxOccurs="unbounded"/>
<xsd:element ref="handler" minOccurs="0" maxOccurs="unbounded"/>
</xsd:sequence>
@@ -140,20 +143,6 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="concurrency">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a concurrency policy.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="core" type="xsd:int"/>
<xsd:attribute name="max" type="xsd:int"/>
<xsd:attribute name="queue-capacity" type="xsd:int"/>
<xsd:attribute name="keep-alive" type="xsd:int"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="selector">
<xsd:complexType>
<xsd:annotation>
@@ -207,4 +196,16 @@
</xsd:complexType>
</xsd:element>
<xsd:complexType name="concurrencyType">
<xsd:annotation>
<xsd:documentation>
Defines a concurrency policy.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="core" type="xsd:int"/>
<xsd:attribute name="max" type="xsd:int"/>
<xsd:attribute name="queue-capacity" type="xsd:int"/>
<xsd:attribute name="keep-alive" type="xsd:int"/>
</xsd:complexType>
</xsd:schema>

19
spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java
View File

@@ -28,6 +28,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.handler.TestHandlers;
import org.springframework.integration.scheduling.Subscription;
@@ -110,4 +111,22 @@ public class MessageBusParserTests {
bus.stop();
}
@Test
public void testDefaultConcurrency() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"messageBusWithDefaultConcurrencyTests.xml", this.getClass());
MessageEndpoint endpoint1 = (MessageEndpoint) context.getBean("endpoint1");
assertEquals(4, endpoint1.getConcurrencyPolicy().getCoreSize());
assertEquals(7, endpoint1.getConcurrencyPolicy().getMaxSize());
}
@Test
public void testExplicitConcurrencyTakesPrecedence() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"messageBusWithDefaultConcurrencyTests.xml", this.getClass());
MessageEndpoint endpoint2 = (MessageEndpoint) context.getBean("endpoint2");
assertEquals(14, endpoint2.getConcurrencyPolicy().getCoreSize());
assertEquals(17, endpoint2.getConcurrencyPolicy().getMaxSize());
}
}

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

@@ -0,0 +1,26 @@
<?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-1.0.xsd">
<message-bus>
<default-concurrency core="4" max="7"/>
</message-bus>
<channel id="channel"/>
<endpoint id="endpoint1" handler-ref="testHandler" input-channel="channel"/>
<endpoint id="endpoint2" handler-ref="testHandler" input-channel="channel">
<concurrency core="14" max="17"/>
</endpoint>
<beans:bean id="testHandler" class="org.springframework.integration.config.TestHandler">
<beans:constructor-arg value="3"/>
</beans:bean>
</beans:beans>