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

properly support request and reply pubSubDomain

Browse Source
This commit is contained in:
Mark Fisher
2011-08-29 17:40:50 -04:00
parent 4d97cb5833
commit 9b6d3cd838
4 changed files with 91 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -67,7 +67,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
private volatile DestinationResolver destinationResolver = new DynamicDestinationResolver();
private volatile boolean pubSubDomain;
private volatile boolean requestPubSubDomain;
private volatile boolean replyPubSubDomain;
private volatile long receiveTimeout = 5000;
@@ -123,7 +125,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
*/
public void setRequestDestination(Destination requestDestination) {
if (requestDestination instanceof Topic) {
this.pubSubDomain = true;
this.requestPubSubDomain = true;
}
this.requestDestination = requestDestination;
}
@@ -141,6 +143,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
* If none is provided, this gateway will create a {@link TemporaryQueue} per invocation.
*/
public void setReplyDestination(Destination replyDestination) {
if (replyDestination instanceof Topic) {
this.replyPubSubDomain = true;
}
this.replyDestination = replyDestination;
}
@@ -166,10 +171,21 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
* necessary when providing a destination name for a Topic rather than
* a destination reference.
*
* @param pubSubDomain true if the request destination is a Topic
* @param requestPubSubDomain true if the request destination is a Topic
*/
public void setPubSubDomain(boolean pubSubDomain) {
this.pubSubDomain = pubSubDomain;
public void setRequestPubSubDomain(boolean requestPubSubDomain) {
this.requestPubSubDomain = requestPubSubDomain;
}
/**
* Specify whether the reply destination is a Topic. This value is
* necessary when providing a destination name for a Topic rather than
* a destination reference.
*
* @param replyPubSubDomain true if the reply destination is a Topic
*/
public void setReplyPubSubDomain(boolean replyPubSubDomain) {
this.replyPubSubDomain = replyPubSubDomain;
}
/**
@@ -295,7 +311,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
Assert.notNull(this.destinationResolver,
"DestinationResolver is required when relying upon the 'requestDestinationName' property.");
return this.destinationResolver.resolveDestinationName(
session, this.requestDestinationName, this.pubSubDomain);
session, this.requestDestinationName, this.requestPubSubDomain);
}
private Destination getReplyDestination(Session session) throws JMSException {
@@ -306,7 +322,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
Assert.notNull(this.destinationResolver,
"DestinationResolver is required when relying upon the 'replyDestinationName' property.");
return this.destinationResolver.resolveDestinationName(
session, this.replyDestinationName, this.pubSubDomain);
session, this.replyDestinationName, this.replyPubSubDomain);
}
return session.createTemporaryQueue();
}
@@ -521,7 +537,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
*/
protected Connection createConnection() throws JMSException {
ConnectionFactory cf = this.connectionFactory;
if (!this.pubSubDomain && cf instanceof QueueConnectionFactory) {
if (!this.requestPubSubDomain && !this.replyPubSubDomain && cf instanceof QueueConnectionFactory) {
return ((QueueConnectionFactory) cf).createQueueConnection();
}
return cf.createConnection();
@@ -537,7 +553,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
* (such as ActiveMQ's <code>org.apache.activemq.pool.PooledConnectionFactory</code>).
*/
protected Session createSession(Connection connection) throws JMSException {
if (!this.pubSubDomain && connection instanceof QueueConnection) {
if (!this.requestPubSubDomain && !this.replyPubSubDomain && connection instanceof QueueConnection) {
return ((QueueConnection) connection).createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
}
return connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

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

@@ -64,7 +64,8 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-request-payload");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-reply-payload");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "pub-sub-domain");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "request-pub-sub-domain");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-pub-sub-domain");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "time-to-live");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "priority");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "explicit-qos-enabled");

32
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2011 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.
@@ -13,9 +13,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jms.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.spy;
@@ -48,11 +50,12 @@ import org.springframework.jms.support.converter.MessageConverter;
/**
* @author Jonas Partner
* @author Oleg Zhurakousky
* @author Mark Fisher
*/
public class JmsOutboundGatewayParserTests {
@Test
public void testWithDelivertPersistentAttribute(){
public void testWithDeliveryPersistentAttribute(){
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"jmsOutboundGatewayWithDeliveryPersistent.xml", this.getClass());
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("jmsGateway");
@@ -109,7 +112,30 @@ public class JmsOutboundGatewayParserTests {
verify(handler, times(1)).handleMessage(Mockito.any(Message.class));
assertEquals("hello", result);
}
@Test
public void gatewayWithDefaultPubSubDomain() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"jmsOutboundGatewayWithPubSubSettings.xml", this.getClass());
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("defaultGateway");
DirectFieldAccessor accessor = new DirectFieldAccessor(
new DirectFieldAccessor(endpoint).getPropertyValue("handler"));
assertFalse((Boolean) accessor.getPropertyValue("requestPubSubDomain"));
assertFalse((Boolean) accessor.getPropertyValue("replyPubSubDomain"));
}
@Test
public void gatewayWithExplicitPubSubDomainTrue() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"jmsOutboundGatewayWithPubSubSettings.xml", this.getClass());
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("pubSubDomainGateway");
DirectFieldAccessor accessor = new DirectFieldAccessor(
new DirectFieldAccessor(endpoint).getPropertyValue("handler"));
assertTrue((Boolean) accessor.getPropertyValue("requestPubSubDomain"));
assertTrue((Boolean) accessor.getPropertyValue("replyPubSubDomain"));
}
public static interface SampleGateway{
public String echo(String value);
}

35
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithPubSubSettings.xml Normal file
View File

@@ -0,0 +1,35 @@
<?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:si="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">
<si:channel id="requestChannel"/>
<jms:outbound-gateway id="defaultGateway"
request-destination-name="requestQueue"
reply-destination-name="replyQueue"
request-channel="requestChannel"/>
<jms:outbound-gateway id="pubSubDomainGateway"
request-destination-name="requestTopic"
reply-destination-name="replyTopic"
request-pub-sub-domain="true"
reply-pub-sub-domain="true"
request-channel="requestChannel"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.jms.StubConnection">
<constructor-arg value="test-message"/>
</bean>
</constructor-arg>
</bean>
</beans>