diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 79433a52a3..94029d0ed8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java index 671f9f3f51..63f5d233e4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java @@ -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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java new file mode 100644 index 0000000000..78886327fb --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java @@ -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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageBusParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageBusParser.java index 4643ede564..cc4c95c1a0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageBusParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageBusParser.java @@ -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); + } + } + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index 175c14b607..2952f6721d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -24,6 +24,9 @@ Defines a message bus. + + + @@ -116,7 +119,7 @@ - + @@ -140,20 +143,6 @@ - - - - - Defines a concurrency policy. - - - - - - - - - @@ -207,4 +196,16 @@ + + + + Defines a concurrency policy. + + + + + + + + \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java index c29be8e56b..5a4061c768 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java @@ -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()); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithDefaultConcurrencyTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithDefaultConcurrencyTests.xml new file mode 100644 index 0000000000..fcbf760dd0 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithDefaultConcurrencyTests.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + + + + + +