From f9741d7ef39d411fab928262877521b00dc73459 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 18 Jun 2012 14:43:53 -0400 Subject: [PATCH] INT-2523 Add QOS Attributes on JMS Channel The Factory Bean already supported the attributes but the namespace did not. Also, the priority was fixed (if set via the factory bean). Add QOS attributes to schema/parser; add use of DynamicJmsTemplateProperties to allow override of configured priority with the priority header (in the same way as the outbound endpoints). --- .../integration/jms/AbstractJmsChannel.java | 11 ++- .../jms/config/JmsAdapterParserUtils.java | 16 ++-- .../jms/config/JmsChannelParser.java | 11 ++- .../jms/config/spring-integration-jms-2.2.xsd | 90 +++++++++++++++++-- .../jms/PollableJmsChannelTests.java | 76 +++++++++++++--- .../config/JmsChannelParserTests-context.xml | 6 +- .../jms/config/JmsChannelParserTests.java | 11 ++- 7 files changed, 187 insertions(+), 34 deletions(-) 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);