diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java index 4a77992c6f..fb9c644733 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -23,6 +23,7 @@ import org.springframework.util.Assert; /** * @author Mark Fisher + * @author Gary Russell * @since 2.0 */ public abstract class AbstractJmsChannel extends AbstractMessageChannel { @@ -42,7 +43,13 @@ public abstract class AbstractJmsChannel extends AbstractMessageChannel { @Override protected boolean doSend(Message message, long timeout) { - this.jmsTemplate.convertAndSend(message); + try { + DynamicJmsTemplateProperties.setPriority(message.getHeaders().getPriority()); + this.jmsTemplate.convertAndSend(message); + } + finally { + DynamicJmsTemplateProperties.clearPriority(); + } return true; } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java index 9b2a5e3a76..f3c522b360 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -16,19 +16,20 @@ package org.springframework.integration.jms.config; -import org.w3c.dom.Element; - import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.jms.DynamicJmsTemplate; import org.springframework.jms.core.JmsTemplate; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; /** * Utility methods and constants for JMS adapter parsers. - * + * * @author Mark Fisher + * @author Gary Russell */ abstract class JmsAdapterParserUtils { @@ -118,8 +119,7 @@ abstract class JmsAdapterParserUtils { } static BeanDefinition parseJmsTemplateBeanDefinition(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.jms.DynamicJmsTemplate"); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(DynamicJmsTemplate.class); builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY, JmsAdapterParserUtils.determineConnectionFactoryBeanName(element, parserContext)); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); @@ -140,7 +140,7 @@ abstract class JmsAdapterParserUtils { if (acknowledgeMode != null) { builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode); } - + return builder.getBeanDefinition(); } @@ -152,5 +152,5 @@ abstract class JmsAdapterParserUtils { } } } - + } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java index 3b03ac35c2..8e898088f5 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java @@ -18,18 +18,17 @@ package org.springframework.integration.jms.config; import javax.jms.Session; -import org.w3c.dom.Element; - import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractChannelParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; /** * Parser for the 'channel' and 'publish-subscribe-channel' elements of the * Spring Integration JMS namespace. - * + * * @author Mark Fisher * @author Oleg Zhurakusky * @author Gary Russell @@ -47,7 +46,7 @@ public class JmsChannelParser extends AbstractChannelParser { @Override protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.jms.config.JmsChannelFactoryBean"); + JmsChannelFactoryBean.class); String messageDriven = element.getAttribute("message-driven"); if (StringUtils.hasText(messageDriven)) { builder.addConstructorArgValue(messageDriven); @@ -90,6 +89,10 @@ public class JmsChannelParser extends AbstractChannelParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "selector", "messageSelector"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delivery-persistent"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "time-to-live"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "priority"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "explicit-qos-enabled"); String cache = element.getAttribute("cache"); if (StringUtils.hasText(cache)) { if (containerType.startsWith("simple")) { diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd index 389303aecd..1fba34b929 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd @@ -342,6 +342,40 @@ ]]> + + + + Specify a boolean value indicating whether the delivery mode should be + DeliveryMode.PERSISTENT (true) or DeliveryMode.NON_PERSISTENT (false). + This setting will only take effect if 'explicit-qos-enabled' is true. + + + + + + + Specify the message time to live. + This setting will only take effect if 'explicit-qos-enabled' is true. + + + + + + + Specify the default priority of the message. Overridden by the message priority + header, if present; range 0-9. + This setting will only take effect if 'explicit-qos-enabled' is true. + + + + + + + + @@ -870,9 +904,31 @@ - - - + + + + Specify the message time to live. + This setting will only take effect if 'explicit-qos-enabled' is true. + + + + + + + Specify the default priority of the message. Overridden by the message priority + header, if present; range 0-9 + This setting will only take effect if 'explicit-qos-enabled' is true. + + + + + + + + @@ -976,9 +1032,31 @@ - - - + + + + Specify the message time to live. + This setting will only take effect if 'explicit-qos-enabled' is true. + + + + + + + Specify the default priority of the message. Overridden by the message priority + header, if present; range 0-9. + This setting will only take effect if 'explicit-qos-enabled' is true. + + + + + + + + ("foo")); + assertTrue(sent1); + final AtomicReference message = new AtomicReference(); + final CountDownLatch latch1 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + message.set(receiver.receive(queue)); + latch1.countDown(); + }}); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertNotNull(message.get()); + assertEquals(5, message.get().getJMSPriority()); + assertTrue(message.get().getJMSExpiration() <= System.currentTimeMillis() + ttl); + assertTrue(message.get().toString().contains("persistent = false")); + message.set(null); + final CountDownLatch latch2 = new CountDownLatch(1); + boolean sent2 = channel.send(MessageBuilder.withPayload("foo").setPriority(6).build()); + assertTrue(sent2); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + message.set(receiver.receive(queue)); + latch2.countDown(); + }}); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertNotNull(message.get()); + assertEquals(6, message.get().getJMSPriority()); + assertTrue(message.get().getJMSExpiration() <= System.currentTimeMillis() + ttl); + assertTrue(message.get().toString().contains("persistent = false")); + } + public static class SampleInterceptor implements ChannelInterceptor { private final boolean preRecieveFlag; public SampleInterceptor(boolean preRecieveFlag){ this.preRecieveFlag = preRecieveFlag; } - + public Message preSend(Message message, MessageChannel channel) { return message; } @@ -167,6 +223,6 @@ public class PollableJmsChannelTests { public Message postReceive(Message message, MessageChannel channel) { return message; } - + } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml index 3060721916..23a9719be5 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml @@ -10,7 +10,11 @@ - + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java index 5210ffa2eb..577227488f 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import java.util.List; +import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Queue; @@ -71,10 +72,10 @@ public class JmsChannelParserTests { @Autowired private MessageChannel topicNameChannel; - + @Autowired private MessageChannel withPlaceholders; - + @Autowired private MessageChannel topicNameWithResolverChannel; @@ -126,6 +127,10 @@ public class JmsChannelParserTests { AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container"); assertEquals(queue, jmsTemplate.getDefaultDestination()); assertEquals(queue, container.getDestination()); + assertEquals(true, TestUtils.getPropertyValue(jmsTemplate, "explicitQosEnabled")); + assertEquals(DeliveryMode.PERSISTENT, TestUtils.getPropertyValue(jmsTemplate, "deliveryMode")); + assertEquals(123L, TestUtils.getPropertyValue(jmsTemplate, "timeToLive")); + assertEquals(12, TestUtils.getPropertyValue(jmsTemplate, "priority")); } @Test @@ -235,7 +240,7 @@ public class JmsChannelParserTests { JmsTemplate jmsTemplate = (JmsTemplate) accessor.getPropertyValue("jmsTemplate"); assertEquals("foo", jmsTemplate.getDefaultDestinationName()); } - + @Test public void withPlaceholders() { DefaultMessageListenerContainer container = TestUtils.getPropertyValue(withPlaceholders, "container", DefaultMessageListenerContainer.class);