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

INT-890 Added QoS attributes (priority, time-to-live, delivery-persistent, and explicit-qos-enabled) to the 'outbound-channel-adapter' element. Also added an 'explicit-qos-enabled' boolean to the 'outbound-gateway' element which now must be set to true to enable the use of its QoS settings.

Browse Source
This commit is contained in:
Mark Fisher
2010-03-12 19:02:14 +00:00
parent 0596b61d6b
commit 1c17ddb121
8 changed files with 146 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/AbstractJmsTemplateBasedAdapter.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.jms;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import org.springframework.beans.factory.InitializingBean;
@@ -41,6 +42,14 @@ public abstract class AbstractJmsTemplateBasedAdapter implements InitializingBea
private volatile DestinationResolver destinationResolver;
private volatile int deliveryMode = javax.jms.Message.DEFAULT_DELIVERY_MODE;
private volatile long timeToLive = javax.jms.Message.DEFAULT_TIME_TO_LIVE;
private volatile int priority = javax.jms.Message.DEFAULT_PRIORITY;
private volatile boolean explicitQosEnabled;
private volatile JmsTemplate jmsTemplate;
private volatile JmsHeaderMapper headerMapper;
@@ -100,6 +109,42 @@ public abstract class AbstractJmsTemplateBasedAdapter implements InitializingBea
this.headerMapper = headerMapper;
}
/**
* @see JmsTemplate#setExplicitQosEnabled(boolean)
*/
public void setExplicitQosEnabled(boolean explicitQosEnabled) {
this.explicitQosEnabled = explicitQosEnabled;
}
/**
* @see JmsTemplate#setTimeToLive(long)
*/
public void setTimeToLive(long timeToLive) {
this.timeToLive = timeToLive;
}
/**
* @see JmsTemplate#setDeliveryMode(int)
*/
public void setDeliveryMode(int deliveryMode) {
this.deliveryMode = deliveryMode;
}
/**
* @see JmsTemplate#setDeliveryPersistent(boolean)
*/
public void setDeliveryPersistent(boolean deliveryPersistent) {
this.deliveryMode = deliveryPersistent ?
DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT;
}
/**
* @see JmsTemplate#setPriority(int)
*/
public void setPriority(int priority) {
this.priority = priority;
}
protected JmsTemplate getJmsTemplate() {
if (this.jmsTemplate == null) {
this.afterPropertiesSet();
@@ -119,6 +164,10 @@ public abstract class AbstractJmsTemplateBasedAdapter implements InitializingBea
+ " 'destination' (or 'destination-name') are required.");
this.jmsTemplate = this.createDefaultJmsTemplate();
}
this.jmsTemplate.setExplicitQosEnabled(this.explicitQosEnabled);
this.jmsTemplate.setTimeToLive(this.timeToLive);
this.jmsTemplate.setPriority(this.priority);
this.jmsTemplate.setDeliveryMode(this.deliveryMode);
this.configureMessageConverter(this.jmsTemplate, this.headerMapper);
this.initialized = true;
}

21
org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -72,6 +72,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
private volatile int priority = javax.jms.Message.DEFAULT_PRIORITY;
private volatile boolean explicitQosEnabled;
private ConnectionFactory connectionFactory;
private volatile MessageConverter messageConverter;
@@ -181,6 +183,14 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
this.timeToLive = timeToLive;
}
/**
* Specify whether explicit QoS settings are enabled
* (deliveryMode, priority, and timeToLive).
*/
public void setExplicitQosEnabled(boolean explicitQosEnabled) {
this.explicitQosEnabled = explicitQosEnabled;
}
/**
* Provide a {@link MessageConverter} strategy to use for converting the
* Spring Integration request Message into a JMS Message and for converting
@@ -312,13 +322,16 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
session = createSession(connection);
javax.jms.Message jmsRequest = this.messageConverter.toMessage(requestMessage, session);
messageProducer = session.createProducer(this.getRequestDestination(session));
messageProducer.setDeliveryMode(this.deliveryMode);
messageProducer.setPriority(this.priority);
messageProducer.setTimeToLive(this.timeToLive);
replyTo = this.getReplyDestination(session);
jmsRequest.setJMSReplyTo(replyTo);
connection.start();
messageProducer.send(jmsRequest);
if (this.explicitQosEnabled) {
messageProducer.send(jmsRequest,
this.deliveryMode, this.priority, this.timeToLive);
}
else {
messageProducer.send(jmsRequest);
}
if (replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic) {
messageConsumer = session.createConsumer(replyTo);
}

2
org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 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.

4
org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java
View File

@@ -71,6 +71,10 @@ public class JmsOutboundChannelAdapterParser extends AbstractOutboundChannelAdap
builder.addPropertyReference(JmsAdapterParserUtils.HEADER_MAPPER_PROPERTY, headerMapper);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "time-to-live");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "priority");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delivery-persistent");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "explicit-qos-enabled");
return builder.getBeanDefinition();
}

3
org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2010 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.
@@ -65,6 +65,7 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delivery-mode");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "time-to-live");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "priority");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "explicit-qos-enabled");
return builder;
}

13
org.springframework.integration.jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd
View File

@@ -633,6 +633,7 @@
<xsd:attribute name="delivery-mode" type="xsd:string"/>
<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="auto-startup" type="xsd:string" default="true"/>
<xsd:attribute name="order" type="xsd:string">
<xsd:annotation>
@@ -701,6 +702,18 @@
</xsd:appinfo>
</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:attribute name="priority" type="xsd:string"/>
<xsd:attribute name="explicit-qos-enabled" type="xsd:string"/>
<xsd:attribute name="order" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[

29
org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 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.
@@ -20,6 +20,8 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import javax.jms.DeliveryMode;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
@@ -97,4 +99,29 @@ public class JmsOutboundChannelAdapterParserTests {
}
}
@Test
public void adapterWithQosSettings() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"jmsOutboundWithQos.xml", this.getClass());
EventDrivenConsumer endpoint = context.getBean("qosAdapter", EventDrivenConsumer.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(
new DirectFieldAccessor(new DirectFieldAccessor(endpoint).getPropertyValue("handler"))
.getPropertyValue("jmsTemplate"));
assertEquals(true, accessor.getPropertyValue("explicitQosEnabled"));
assertEquals(12345L, accessor.getPropertyValue("timeToLive"));
assertEquals(7, accessor.getPropertyValue("priority"));
assertEquals(DeliveryMode.NON_PERSISTENT, accessor.getPropertyValue("deliveryMode"));
}
@Test
public void qosNotExplicitByDefault() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"jmsOutboundWithQos.xml", this.getClass());
EventDrivenConsumer endpoint = context.getBean("defaultAdapter", EventDrivenConsumer.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(
new DirectFieldAccessor(new DirectFieldAccessor(endpoint).getPropertyValue("handler"))
.getPropertyValue("jmsTemplate"));
assertEquals(false, accessor.getPropertyValue("explicitQosEnabled"));
}
}

32
org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundWithQos.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:integration="http://www.springframework.org/schema/integration"
xmlns:jms="http://www.springframework.org/schema/integration/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms
http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
<integration:channel id="input"/>
<jms:outbound-channel-adapter id="qosAdapter"
channel="input" destination="testDestination" explicit-qos-enabled="true"
time-to-live="12345" priority="7" delivery-persistent="false"/>
<jms:outbound-channel-adapter id="defaultAdapter"
channel="input" destination="testDestination"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.jms.StubConnection">
<constructor-arg value="target-test"/>
</bean>
</constructor-arg>
</bean>
<bean id="testDestination" class="org.springframework.integration.jms.StubDestination"/>
</beans>