diff --git a/spring-jms/src/main/java/org/springframework/jms/annotation/JmsListener.java b/spring-jms/src/main/java/org/springframework/jms/annotation/JmsListener.java index 375029c0bd..bd82b1855c 100644 --- a/spring-jms/src/main/java/org/springframework/jms/annotation/JmsListener.java +++ b/spring-jms/src/main/java/org/springframework/jms/annotation/JmsListener.java @@ -107,4 +107,13 @@ public @interface JmsListener { */ String selector() default ""; + /** + * The concurrency for the listener, if any. + *

The concurrency limits can be a "lower-upper" String, e.g. "5-10", or a simple + * upper limit String, e.g. "10" (the lower limit will be 1 in this case). + *

The underlying container may or may not support all features. For instance, it + * may not be able to scale: in that case only the upper value is used. + */ + String concurrency() default ""; + } diff --git a/spring-jms/src/main/java/org/springframework/jms/annotation/JmsListenerAnnotationBeanPostProcessor.java b/spring-jms/src/main/java/org/springframework/jms/annotation/JmsListenerAnnotationBeanPostProcessor.java index 872589f69c..9e1628b72e 100644 --- a/spring-jms/src/main/java/org/springframework/jms/annotation/JmsListenerAnnotationBeanPostProcessor.java +++ b/spring-jms/src/main/java/org/springframework/jms/annotation/JmsListenerAnnotationBeanPostProcessor.java @@ -181,6 +181,9 @@ public class JmsListenerAnnotationBeanPostProcessor implements BeanPostProcessor if (StringUtils.hasText(jmsListener.subscription())) { endpoint.setSubscription(jmsListener.subscription()); } + if (StringUtils.hasText(jmsListener.concurrency())) { + endpoint.setConcurrency(jmsListener.concurrency()); + } JmsListenerContainerFactory factory = null; String containerFactoryBeanName = jmsListener.containerFactory(); diff --git a/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerEndpoint.java b/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerEndpoint.java index a355984b57..7cf0909ee6 100644 --- a/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerEndpoint.java +++ b/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerEndpoint.java @@ -42,6 +42,8 @@ public abstract class AbstractJmsListenerEndpoint implements JmsListenerEndpoint private String selector; + private String concurrency; + public void setId(String id) { this.id = id; @@ -95,6 +97,23 @@ public abstract class AbstractJmsListenerEndpoint implements JmsListenerEndpoint return this.selector; } + /** + * Set a concurrency for the listener, if any. + *

The concurrency limits can be a "lower-upper" String, e.g. "5-10", or a simple + * upper limit String, e.g. "10" (the lower limit will be 1 in this case). + *

The underlying container may or may not support all features. For instance, it + * may not be able to scale: in that case only the upper value is used. + */ + public void setConcurrency(String concurrency) { + this.concurrency = concurrency; + } + + /** + * Return the concurrency for the listener, if any. + */ + public String getConcurrency() { + return concurrency; + } @Override public void setupMessageContainer(MessageListenerContainer container) { @@ -121,6 +140,9 @@ public abstract class AbstractJmsListenerEndpoint implements JmsListenerEndpoint if (getSelector() != null) { container.setMessageSelector(getSelector()); } + if (getConcurrency() != null) { + container.setConcurrency(getConcurrency()); + } setupMessageListener(container); } @@ -140,6 +162,9 @@ public abstract class AbstractJmsListenerEndpoint implements JmsListenerEndpoint if (getSelector() != null) { activationSpecConfig.setMessageSelector(getSelector()); } + if (getConcurrency() != null) { + activationSpecConfig.setConcurrency(getConcurrency()); + } setupMessageListener(container); } diff --git a/spring-jms/src/main/java/org/springframework/jms/config/AbstractListenerContainerParser.java b/spring-jms/src/main/java/org/springframework/jms/config/AbstractListenerContainerParser.java index ccb23c68c4..9384de78eb 100644 --- a/spring-jms/src/main/java/org/springframework/jms/config/AbstractListenerContainerParser.java +++ b/spring-jms/src/main/java/org/springframework/jms/config/AbstractListenerContainerParser.java @@ -247,6 +247,15 @@ abstract class AbstractListenerContainerParser implements BeanDefinitionParser { } configDef.getPropertyValues().add("messageSelector", selector); } + + if (ele.hasAttribute(CONCURRENCY_ATTRIBUTE)) { + String concurrency = ele.getAttribute(CONCURRENCY_ATTRIBUTE); + if (!StringUtils.hasText(concurrency)) { + parserContext.getReaderContext().error( + "Listener 'concurrency' attribute contains empty value.", ele); + } + configDef.getPropertyValues().add("concurrency", concurrency); + } } protected PropertyValues parseCommonContainerProperties(Element ele, ParserContext parserContext) { diff --git a/spring-jms/src/main/java/org/springframework/jms/config/JcaListenerContainerParser.java b/spring-jms/src/main/java/org/springframework/jms/config/JcaListenerContainerParser.java index 8bfff85455..c46a8d3e04 100644 --- a/spring-jms/src/main/java/org/springframework/jms/config/JcaListenerContainerParser.java +++ b/spring-jms/src/main/java/org/springframework/jms/config/JcaListenerContainerParser.java @@ -70,10 +70,10 @@ class JcaListenerContainerParser extends AbstractListenerContainerParser { containerDef.setSource(context.getSource()); containerDef.setBeanClassName("org.springframework.jms.listener.endpoint.JmsMessageEndpointManager"); - containerDef.getPropertyValues().addPropertyValues(context.getContainerValues()); + applyContainerValues(context, containerDef); - BeanDefinition activationSpec = getActivationSpecConfigBeanDefinition(context.getContainerValues()); + BeanDefinition activationSpec = getActivationSpecConfigBeanDefinition(containerDef.getPropertyValues()); parseListenerConfiguration(context.getListenerElement(), context.getParserContext(), activationSpec); String phase = context.getContainerElement().getAttribute(PHASE_ATTRIBUTE); @@ -84,6 +84,21 @@ class JcaListenerContainerParser extends AbstractListenerContainerParser { return containerDef; } + /** + * The property values provided by the factory element contains a mutable property (the + * activation spec config). To avoid changing the bean definition from the parent, a clone + * bean definition is created for the container being configured. + */ + private void applyContainerValues(ListenerContainerParserContext context, RootBeanDefinition containerDef) { + // Apply settings from the container + containerDef.getPropertyValues().addPropertyValues(context.getContainerValues()); + + // Clone the activationSpecConfig property value as it is mutable + PropertyValue pv = containerDef.getPropertyValues().getPropertyValue("activationSpecConfig"); + RootBeanDefinition activationSpecConfig = new RootBeanDefinition((RootBeanDefinition) pv.getValue()); + containerDef.getPropertyValues().add("activationSpecConfig", activationSpecConfig); + } + @Override protected boolean indicatesPubSub(PropertyValues propertyValues) { BeanDefinition configDef = getActivationSpecConfigBeanDefinition(propertyValues); diff --git a/spring-jms/src/main/java/org/springframework/jms/config/JmsListenerContainerParser.java b/spring-jms/src/main/java/org/springframework/jms/config/JmsListenerContainerParser.java index b3a226f156..38eacd2584 100644 --- a/spring-jms/src/main/java/org/springframework/jms/config/JmsListenerContainerParser.java +++ b/spring-jms/src/main/java/org/springframework/jms/config/JmsListenerContainerParser.java @@ -92,7 +92,6 @@ class JmsListenerContainerParser extends AbstractListenerContainerParser { // Set all container values containerDef.getPropertyValues().addPropertyValues(context.getContainerValues()); - parseListenerConfiguration(context.getListenerElement(), context.getParserContext(), containerDef); Element containerEle = context.getContainerElement(); String containerType = containerEle.getAttribute(CONTAINER_TYPE_ATTRIBUTE); @@ -116,6 +115,9 @@ class JmsListenerContainerParser extends AbstractListenerContainerParser { containerDef.getPropertyValues().add("phase", phase); } + // Parse listener specific settings + parseListenerConfiguration(context.getListenerElement(), context.getParserContext(), containerDef); + return containerDef; } diff --git a/spring-jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java b/spring-jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java index 36a5ce6c4a..e74dbe6912 100644 --- a/spring-jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java +++ b/spring-jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java @@ -147,6 +147,11 @@ public abstract class AbstractMessageListenerContainer private boolean acceptMessagesWhileStopping = false; + /** + * Specify concurrency limits. + */ + public abstract void setConcurrency(String concurrency); + /** * Set the destination to receive messages from. *

Alternatively, specify a "destinationName", to be dynamically diff --git a/spring-jms/src/main/java/org/springframework/jms/listener/DefaultMessageListenerContainer.java b/spring-jms/src/main/java/org/springframework/jms/listener/DefaultMessageListenerContainer.java index 4c942f512e..7eb1d28610 100644 --- a/spring-jms/src/main/java/org/springframework/jms/listener/DefaultMessageListenerContainer.java +++ b/spring-jms/src/main/java/org/springframework/jms/listener/DefaultMessageListenerContainer.java @@ -295,6 +295,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe * ({@link #setConcurrentConsumers}) and will slowly scale up to the maximum number * of consumers {@link #setMaxConcurrentConsumers} in case of increasing load. */ + @Override public void setConcurrency(String concurrency) { try { int separatorIndex = concurrency.indexOf('-'); diff --git a/spring-jms/src/main/java/org/springframework/jms/listener/SimpleMessageListenerContainer.java b/spring-jms/src/main/java/org/springframework/jms/listener/SimpleMessageListenerContainer.java index 2a503cc754..7f013082ba 100644 --- a/spring-jms/src/main/java/org/springframework/jms/listener/SimpleMessageListenerContainer.java +++ b/spring-jms/src/main/java/org/springframework/jms/listener/SimpleMessageListenerContainer.java @@ -114,6 +114,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta * {@link DefaultMessageListenerContainer}. For this local listener container, * generally use {@link #setConcurrentConsumers} instead. */ + @Override public void setConcurrency(String concurrency) { try { int separatorIndex = concurrency.indexOf('-'); diff --git a/spring-jms/src/main/resources/org/springframework/jms/config/spring-jms-4.1.xsd b/spring-jms/src/main/resources/org/springframework/jms/config/spring-jms-4.1.xsd index 6df309c198..8c26ec77eb 100644 --- a/spring-jms/src/main/resources/org/springframework/jms/config/spring-jms-4.1.xsd +++ b/spring-jms/src/main/resources/org/springframework/jms/config/spring-jms-4.1.xsd @@ -590,6 +590,17 @@ ]]> + + + + + diff --git a/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerEndpointTests.java b/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerEndpointTests.java index cc591c1ccc..3e1780093b 100644 --- a/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerEndpointTests.java +++ b/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerEndpointTests.java @@ -25,8 +25,10 @@ import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.jms.listener.MessageListenerContainer; +import org.springframework.jms.listener.SimpleMessageListenerContainer; import org.springframework.jms.listener.adapter.MessageListenerAdapter; import org.springframework.jms.listener.endpoint.JmsActivationSpecConfig; import org.springframework.jms.listener.endpoint.JmsMessageEndpointManager; @@ -48,12 +50,15 @@ public class JmsListenerEndpointTests { endpoint.setDestination("myQueue"); endpoint.setSelector("foo = 'bar'"); endpoint.setSubscription("mySubscription"); + endpoint.setConcurrency("5-10"); endpoint.setMessageListener(messageListener); endpoint.setupMessageContainer(container); assertEquals("myQueue", container.getDestinationName()); assertEquals("foo = 'bar'", container.getMessageSelector()); assertEquals("mySubscription", container.getDurableSubscriptionName()); + assertEquals(5, container.getConcurrentConsumers()); + assertEquals(10, container.getMaxConcurrentConsumers()); assertEquals(messageListener, container.getMessageListener()); } @@ -65,6 +70,7 @@ public class JmsListenerEndpointTests { endpoint.setDestination("myQueue"); endpoint.setSelector("foo = 'bar'"); endpoint.setSubscription("mySubscription"); + endpoint.setConcurrency("10"); endpoint.setMessageListener(messageListener); endpoint.setupMessageContainer(container); @@ -72,9 +78,21 @@ public class JmsListenerEndpointTests { assertEquals("myQueue", config.getDestinationName()); assertEquals("foo = 'bar'", config.getMessageSelector()); assertEquals("mySubscription", config.getDurableSubscriptionName()); + assertEquals(10, config.getMaxConcurrency()); assertEquals(messageListener, container.getMessageListener()); } + @Test + public void setupConcurrencySimpleContainer() { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + MessageListener messageListener = new MessageListenerAdapter(); + SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint(); + endpoint.setConcurrency("5-10"); // simple implementation only support max value + endpoint.setMessageListener(messageListener); + + endpoint.setupMessageContainer(container); + assertEquals(10, new DirectFieldAccessor(container).getPropertyValue("concurrentConsumers")); + } @Test public void setupMessageContainerNoListener() { diff --git a/spring-jms/src/test/java/org/springframework/jms/config/JmsNamespaceHandlerTests.java b/spring-jms/src/test/java/org/springframework/jms/config/JmsNamespaceHandlerTests.java index d14169883c..8d24ee0ab5 100644 --- a/spring-jms/src/test/java/org/springframework/jms/config/JmsNamespaceHandlerTests.java +++ b/spring-jms/src/test/java/org/springframework/jms/config/JmsNamespaceHandlerTests.java @@ -102,13 +102,9 @@ public class JmsNamespaceHandlerTests { for (DefaultMessageListenerContainer container : containers.values()) { if (container.getConnectionFactory().equals(defaultConnectionFactory)) { defaultConnectionFactoryCount++; - assertEquals(2, container.getConcurrentConsumers()); - assertEquals(3, container.getMaxConcurrentConsumers()); } else if (container.getConnectionFactory().equals(explicitConnectionFactory)) { explicitConnectionFactoryCount++; - assertEquals(3, container.getConcurrentConsumers()); - assertEquals(5, container.getMaxConcurrentConsumers()); } } @@ -226,6 +222,34 @@ public class JmsNamespaceHandlerTests { assertEquals(DefaultMessageListenerContainer.DEFAULT_RECOVERY_INTERVAL, recoveryInterval3); } + @Test + public void testConcurrency() { + // JMS + DefaultMessageListenerContainer listener0 = this.context + .getBean(DefaultMessageListenerContainer.class.getName() + "#0", DefaultMessageListenerContainer.class); + DefaultMessageListenerContainer listener1 = this.context + .getBean("listener1", DefaultMessageListenerContainer.class); + DefaultMessageListenerContainer listener2 = this.context + .getBean("listener2", DefaultMessageListenerContainer.class); + + assertEquals("Wrong concurrency on listener using placeholder", 2, listener0.getConcurrentConsumers()); + assertEquals("Wrong concurrency on listener using placeholder", 3, listener0.getMaxConcurrentConsumers()); + assertEquals("Wrong concurrency on listener1", 3, listener1.getConcurrentConsumers()); + assertEquals("Wrong max concurrency on listener1", 5, listener1.getMaxConcurrentConsumers()); + assertEquals("Wrong custom concurrency on listener2", 5, listener2.getConcurrentConsumers()); + assertEquals("Wrong custom max concurrency on listener2", 10, listener2.getMaxConcurrentConsumers()); + + // JCA + JmsMessageEndpointManager listener3 = this.context + .getBean("listener3", JmsMessageEndpointManager.class); + JmsMessageEndpointManager listener4 = this.context + .getBean("listener4", JmsMessageEndpointManager.class); + assertEquals("Wrong concurrency on listener3", 5, + listener3.getActivationSpecConfig().getMaxConcurrency()); + assertEquals("Wrong custom concurrency on listener4", 7, + listener4.getActivationSpecConfig().getMaxConcurrency()); + } + @Test public void testErrorHandlers() { ErrorHandler expected = this.context.getBean("testErrorHandler", ErrorHandler.class); diff --git a/spring-jms/src/test/resources/org/springframework/jms/config/jmsNamespaceHandlerTests.xml b/spring-jms/src/test/resources/org/springframework/jms/config/jmsNamespaceHandlerTests.xml index e5bad55eb4..1e17a06329 100644 --- a/spring-jms/src/test/resources/org/springframework/jms/config/jmsNamespaceHandlerTests.xml +++ b/spring-jms/src/test/resources/org/springframework/jms/config/jmsNamespaceHandlerTests.xml @@ -11,7 +11,8 @@ transaction-manager="testTransactionManager" error-handler="testErrorHandler" cache="connection" concurrency="3-5" prefetch="50" receive-timeout="100" back-off="testBackOff" phase="99"> - + @@ -31,7 +32,8 @@ resource-adapter="testResourceAdapter" activation-spec-factory="testActivationSpecFactory" message-converter="testMessageConverter" concurrency="5" prefetch="50" phase="77"> - + diff --git a/src/asciidoc/index.adoc b/src/asciidoc/index.adoc index f35a9f6361..41aef153d5 100644 --- a/src/asciidoc/index.adoc +++ b/src/asciidoc/index.adoc @@ -41661,6 +41661,12 @@ describes all available attributes: | selector | An optional message selector for this listener. + +| concurrency +| The number of concurrent sessions/consumers to start for this listener. Can either be +| a simple number indicating the maximum number (e.g. "5") or a range indicating the lower +| as well as the upper limit (e.g. "3-5"). Note that a specified minimum is just a hint +| and might be ignored at runtime. Default is the value provided by the container |=== The `` element also accepts several optional attributes. This