Support custom resolution of response destination

Previously, the "pubSubDomain" drove the resolution of both the
destination of the listener and the default response destination.

A new "replyPubSubDomain" attribute has been added on the base listener
and  can be used to listen on a topic and reply to a queue (or vice
versa). The attribute is exposed via the "response-destination-type" XML
attribute on the listener container element. It is also available on the
JmsListenerContainerFactory for use with the @JmsListener infrastructure.

Issue: SPR-12911

Author: snicoll

spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java

@@ -52,6 +52,8 @@ public abstract class AbstractJmsListenerContainerFactory<C extends AbstractMess
 
 	private Boolean pubSubDomain;
 
+	private Boolean replyPubSubDomain;
+
 	private Boolean subscriptionDurable;
 
 	private Boolean subscriptionShared;
@@ -112,6 +114,13 @@ public void setPubSubDomain(Boolean pubSubDomain) {
 		this.pubSubDomain = pubSubDomain;
 	}
 
+	/**
+	 * @see AbstractMessageListenerContainer#setReplyPubSubDomain(boolean)
+	 */
+	public void setReplyPubSubDomain(Boolean replyPubSubDomain) {
+		this.replyPubSubDomain = replyPubSubDomain;
+	}
+
 	/**
 	 * @see AbstractMessageListenerContainer#setSubscriptionDurable(boolean)
 	 */
@@ -172,6 +181,9 @@ public C createListenerContainer(JmsListenerEndpoint endpoint) {
 		if (this.pubSubDomain != null) {
 			instance.setPubSubDomain(this.pubSubDomain);
 		}
+		if (this.replyPubSubDomain != null) {
+			instance.setReplyPubSubDomain(this.replyPubSubDomain);
+		}
 		if (this.subscriptionDurable != null) {
 			instance.setSubscriptionDurable(this.subscriptionDurable);
 		}

spring-jms/src/main/java/org/springframework/jms/config/AbstractListenerContainerParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -78,6 +78,8 @@ abstract class AbstractListenerContainerParser implements BeanDefinitionParser {
 
 	protected static final String DESTINATION_TYPE_SHARED_DURABLE_TOPIC = "sharedDurableTopic";
 
+	protected static final String RESPONSE_DESTINATION_TYPE_ATTRIBUTE = "response-destination-type";
+
 	protected static final String CLIENT_ID_ATTRIBUTE = "client-id";
 
 	protected static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge";
@@ -170,7 +172,7 @@ private void parseListener(Element containerEle, Element listenerEle, ParserCont
 
 		if (listenerEle.hasAttribute(RESPONSE_DESTINATION_ATTRIBUTE)) {
 			String responseDestination = listenerEle.getAttribute(RESPONSE_DESTINATION_ATTRIBUTE);
-			Boolean pubSubDomain = (Boolean) commonContainerProperties.getPropertyValue("pubSubDomain").getValue();
+			Boolean pubSubDomain = (Boolean) commonContainerProperties.getPropertyValue("replyPubSubDomain").getValue();
 			listenerDef.getPropertyValues().add(
 					pubSubDomain ? "defaultResponseTopicName" : "defaultResponseQueueName", responseDestination);
 			if (containerDef.getPropertyValues().contains("destinationResolver")) {
@@ -260,6 +262,23 @@ else if ("".equals(destinationType) || DESTINATION_TYPE_QUEUE.equals(destination
 		properties.add("subscriptionDurable", subscriptionDurable);
 		properties.add("subscriptionShared", subscriptionShared);
 
+		boolean replyPubSubDomain = false;
+		String replyDestinationType = containerEle.getAttribute(RESPONSE_DESTINATION_TYPE_ATTRIBUTE);
+		if (DESTINATION_TYPE_TOPIC.equals(replyDestinationType)) {
+			replyPubSubDomain = true;
+		}
+		else if (DESTINATION_TYPE_QUEUE.equals(replyDestinationType)) {
+			replyPubSubDomain = false;
+		}
+		else if (!StringUtils.hasText(replyDestinationType)) {
+			replyPubSubDomain = pubSubDomain; // the default: same value as pubSubDomain
+		}
+		else if (StringUtils.hasText(replyDestinationType)) {
+			parserContext.getReaderContext().error("Invalid listener container 'response-destination-type': only " +
+					"\"queue\", \"topic\" supported.", containerEle);
+		}
+		properties.add("replyPubSubDomain", replyPubSubDomain);
+
 		if (containerEle.hasAttribute(CLIENT_ID_ATTRIBUTE)) {
 			String clientId = containerEle.getAttribute(CLIENT_ID_ATTRIBUTE);
 			if (!StringUtils.hasText(clientId)) {

spring-jms/src/main/java/org/springframework/jms/config/MethodJmsListenerEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -89,7 +89,7 @@ protected MessagingMessageListenerAdapter createMessageListener(MessageListenerC
 		messageListener.setHandlerMethod(invocableHandlerMethod);
 		String responseDestination = getDefaultResponseDestination();
 		if (StringUtils.hasText(responseDestination)) {
-			if (container.isPubSubDomain()) {
+			if (container.isReplyPubSubDomain()) {
 				messageListener.setDefaultResponseTopicName(responseDestination);
 			}
 			else {

spring-jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java

@@ -153,6 +153,8 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 
 	private String subscriptionName;
 
+	private Boolean replyPubSubDomain;
+
 	private boolean pubSubNoLocal = false;
 
 	private MessageConverter messageConverter;
@@ -445,6 +447,34 @@ public boolean isPubSubNoLocal() {
 		return this.pubSubNoLocal;
 	}
 
+	/**
+	 * Configure the reply destination type. By default, the configured {@code pubSubDomain}
+	 * value is used (see {@link #isPubSubDomain()}.
+	 * <p>This setting primarily indicates what type of destination to resolve
+	 * if dynamic destinations are enabled.
+	 * @param replyPubSubDomain "true" for the Publish/Subscribe domain ({@link javax.jms.Topic Topics}),
+	 * "false" for the Point-to-Point domain ({@link javax.jms.Queue Queues})
+	 * @see #setDestinationResolver
+	 */
+	public void setReplyPubSubDomain(boolean replyPubSubDomain) {
+		this.replyPubSubDomain = replyPubSubDomain;
+	}
+
+	/**
+	 * Return whether the Publish/Subscribe domain ({@link javax.jms.Topic Topics}) is used
+	 * for replies. Otherwise, the Point-to-Point domain ({@link javax.jms.Queue Queues}) is
+	 * used.
+	 */
+	@Override
+	public boolean isReplyPubSubDomain() {
+		if (this.replyPubSubDomain != null) {
+			return replyPubSubDomain;
+		}
+		else {
+			return isPubSubDomain();
+		}
+	}
+
 	/**
 	 * Set the {@link MessageConverter} strategy for converting JMS Messages.
 	 * @since 4.1

spring-jms/src/main/java/org/springframework/jms/listener/MessageListenerContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -47,4 +47,12 @@ public interface MessageListenerContainer extends SmartLifecycle {
 	 */
 	boolean isPubSubDomain();
 
+	/**
+	 * Return whether the reply destination uses Publish/Subscribe domain
+	 * ({@link javax.jms.Topic Topics}). Otherwise, the Point-to-Point domain
+	 * ({@link javax.jms.Queue Queues}) is used.
+	 * <p>By default, the value is identical to {@link #isPubSubDomain()}.
+	 */
+	boolean isReplyPubSubDomain();
+
 }

spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsActivationSpecConfig.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -46,6 +46,8 @@ public class JmsActivationSpecConfig {
 
 	private boolean pubSubDomain = false;
 
+	private Boolean replyPubSubDomain;
+
 	private boolean subscriptionDurable = false;
 
 	private boolean subscriptionShared = false;
@@ -81,6 +83,19 @@ public boolean isPubSubDomain() {
 		return this.pubSubDomain;
 	}
 
+	public void setReplyPubSubDomain(boolean replyPubSubDomain) {
+		this.replyPubSubDomain = replyPubSubDomain;
+	}
+
+	public boolean isReplyPubSubDomain() {
+		if (this.replyPubSubDomain != null) {
+			return this.replyPubSubDomain;
+		}
+		else {
+			return isPubSubDomain();
+		}
+	}
+
 	public void setSubscriptionDurable(boolean subscriptionDurable) {
 		this.subscriptionDurable = subscriptionDurable;
 		if (subscriptionDurable) {

spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsMessageEndpointManager.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -200,4 +200,13 @@ public boolean isPubSubDomain() {
 		throw new IllegalStateException("Could not determine pubSubDomain - no activation spec config is set");
 	}
 
+	@Override
+	public boolean isReplyPubSubDomain() {
+		JmsActivationSpecConfig config = getActivationSpecConfig();
+		if (config != null) {
+			return config.isReplyPubSubDomain();
+		}
+		throw new IllegalStateException("Could not determine reply pubSubDomain - no activation spec config is set");
+	}
+
 }

spring-jms/src/main/resources/META-INF/spring.schemas

@@ -4,4 +4,5 @@ http\://www.springframework.org/schema/jms/spring-jms-3.1.xsd=org/springframewor
 http\://www.springframework.org/schema/jms/spring-jms-3.2.xsd=org/springframework/jms/config/spring-jms-3.2.xsd
 http\://www.springframework.org/schema/jms/spring-jms-4.0.xsd=org/springframework/jms/config/spring-jms-4.0.xsd
 http\://www.springframework.org/schema/jms/spring-jms-4.1.xsd=org/springframework/jms/config/spring-jms-4.1.xsd
-http\://www.springframework.org/schema/jms/spring-jms.xsd=org/springframework/jms/config/spring-jms-4.1.xsd
+http\://www.springframework.org/schema/jms/spring-jms-4.2.xsd=org/springframework/jms/config/spring-jms-4.2.xsd
+http\://www.springframework.org/schema/jms/spring-jms.xsd=org/springframework/jms/config/spring-jms-4.2.xsd