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

Merge pull request #502 from garyrussell/INT-2523

Browse Source 
* garyrussell-INT-2523:
  INT-2523 Add QOS Attributes on JMS Channel
This commit is contained in:
Gunnar Hillert
2012-06-19 11:45:11 -04:00
parent 39491a1df7 f9741d7ef3
commit eead196808
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);