diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 171a3f4550..c10fa186a9 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -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 org.apache.activemq.pool.PooledConnectionFactory). */ 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); diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java index 3d1c8964ec..4206c4269b 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java @@ -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"); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java index ab6725b134..3702c1af56 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java @@ -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); } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithPubSubSettings.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithPubSubSettings.xml new file mode 100644 index 0000000000..ea1d6c74f2 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithPubSubSettings.xml @@ -0,0 +1,35 @@ + + + + + + + + + + + + + + + + + +