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

INT-2523 Add QOS Attributes on JMS Channel

Browse Source 
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).
This commit is contained in:
Gary Russell
2012-06-18 14:43:53 -04:00
committed by Gunnar Hillert
parent 39491a1df7
commit f9741d7ef3
7 changed files with 187 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java
View File

@@ -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;
}

16
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java
View File

@@ -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 {
}
}
}
}

11
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java
View File

@@ -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")) {

90
spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd
View File

@@ -342,6 +342,40 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="delivery-persistent" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="time-to-live" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify the message time to live.
This setting will only take effect if 'explicit-qos-enabled' is true.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="priority" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="explicit-qos-enabled" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Setting this attribute to true enables the use of quality of service attributes - message
priority, delivery mode, time to live.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:element name="message-driven-channel-adapter">
@@ -870,9 +904,31 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="time-to-live" type="xsd:string"/>
<xsd:attribute name="priority" type="xsd:string"/>
<xsd:attribute name="explicit-qos-enabled" type="xsd:string"/>
<xsd:attribute name="time-to-live" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify the message time to live.
This setting will only take effect if 'explicit-qos-enabled' is true.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="priority" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="explicit-qos-enabled" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Setting this attribute to true enables the use of quality of service attributes - message
priority, delivery mode, time to live.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
<xsd:attribute name="order" type="xsd:string">
<xsd:annotation>
@@ -976,9 +1032,31 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="time-to-live" type="xsd:string"/>
<xsd:attribute name="priority" type="xsd:string"/>
<xsd:attribute name="explicit-qos-enabled" type="xsd:string"/>
<xsd:attribute name="time-to-live" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify the message time to live.
This setting will only take effect if 'explicit-qos-enabled' is true.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="priority" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="explicit-qos-enabled" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Setting this attribute to true enables the use of quality of service attributes - message
priority, delivery mode, time to live.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[

76
spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java
View File

@@ -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.
@@ -26,6 +26,10 @@ import static org.mockito.Mockito.verify;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.jms.Destination;
@@ -33,16 +37,19 @@ import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.jms.config.JmsChannelFactoryBean;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
/**
* @author Mark Fisher
* @author Gary Russell
*/
public class PollableJmsChannelTests {
@@ -56,7 +63,7 @@ public class PollableJmsChannelTests {
this.connectionFactory = new ActiveMQConnectionFactory();
this.connectionFactory.setBrokerURL("vm://localhost");
this.queue = new ActiveMQQueue("pollableJmsChannelTestQueue");
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setDestination(this.queue);
@@ -79,7 +86,7 @@ public class PollableJmsChannelTests {
ActiveMqTestUtils.prepare();
this.connectionFactory = new ActiveMQConnectionFactory();
this.connectionFactory.setBrokerURL("vm://localhost");
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setDestinationName("someDynamicQueue");
@@ -97,13 +104,13 @@ public class PollableJmsChannelTests {
assertNotNull(result2);
assertEquals("bar", result2.getPayload());
}
@Test
public void queueNameWithFalsePreReceiveInterceptors() throws Exception {
ActiveMqTestUtils.prepare();
this.connectionFactory = new ActiveMQConnectionFactory();
this.connectionFactory.setBrokerURL("vm://localhost");
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setDestinationName("someDynamicQueue");
@@ -121,13 +128,13 @@ public class PollableJmsChannelTests {
verify(interceptor, times(1)).preReceive(Mockito.any(MessageChannel.class));
verify(interceptor, times(0)).postReceive(Mockito.any(Message.class), Mockito.any(MessageChannel.class));
}
@Test
public void queueNameWithTruePreReceiveInterceptors() throws Exception {
ActiveMqTestUtils.prepare();
this.connectionFactory = new ActiveMQConnectionFactory();
this.connectionFactory.setBrokerURL("vm://localhost");
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
factoryBean.setConnectionFactory(this.connectionFactory);
factoryBean.setDestinationName("someDynamicQueue");
@@ -146,12 +153,61 @@ public class PollableJmsChannelTests {
verify(interceptor, times(1)).postReceive(Mockito.any(Message.class), Mockito.any(MessageChannel.class));
}
@Test
public void qos() throws Exception {
ActiveMqTestUtils.prepare();
this.connectionFactory = new ActiveMQConnectionFactory();
this.connectionFactory.setBrokerURL("vm://localhost");
this.queue = new ActiveMQQueue("pollableJmsChannelTestQueue");
CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
factoryBean.setConnectionFactory(ccf);
factoryBean.setDestination(this.queue);
factoryBean.setExplicitQosEnabled(true);
factoryBean.setPriority(5);
int ttl = 10000;
factoryBean.setTimeToLive(ttl);
factoryBean.setDeliveryPersistent(false);
factoryBean.afterPropertiesSet();
PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject();
final JmsTemplate receiver = new JmsTemplate(this.connectionFactory);
boolean sent1 = channel.send(new GenericMessage<String>("foo"));
assertTrue(sent1);
final AtomicReference<javax.jms.Message> message = new AtomicReference<javax.jms.Message>();
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;
}
}
}

6
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml
View File

@@ -10,7 +10,11 @@
<context:property-placeholder location="classpath:org/springframework/integration/jms/config/channel.properties"/>
<jms:channel id="queueReferenceChannel" queue="testQueue"/>
<jms:channel id="queueReferenceChannel" queue="testQueue"
explicit-qos-enabled="true"
delivery-persistent="true"
time-to-live="123"
priority="12"/>
<jms:channel id="queueNameChannel" queue-name="test.queue"/>

11
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java
View File

@@ -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);