Customize QosSettings for JMS replies

This commit introduces QosSettings that gather the Qualify of Service
settings one can use when sending a message. Such object can now be
associated to any JMS endpoint that allows to send a reply as part of
the processing of an incoming message.

Issue: SPR-15408

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -22,6 +22,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.support.QosSettings;
 import org.springframework.jms.support.converter.MessageConverter;
 import org.springframework.jms.support.destination.DestinationResolver;
 import org.springframework.util.ErrorHandler;
@@ -54,6 +55,8 @@ public abstract class AbstractJmsListenerContainerFactory<C extends AbstractMess
 
 	private Boolean replyPubSubDomain;
 
+	private QosSettings replyQosSettings;
+
 	private Boolean subscriptionDurable;
 
 	private Boolean subscriptionShared;
@@ -121,6 +124,13 @@ public abstract class AbstractJmsListenerContainerFactory<C extends AbstractMess
 		this.replyPubSubDomain = replyPubSubDomain;
 	}
 
+	/**
+	 * @see AbstractMessageListenerContainer#setReplyQosSettings(QosSettings)
+	 */
+	public void setReplyQosSettings(QosSettings replyQosSettings) {
+		this.replyQosSettings = replyQosSettings;
+	}
+
 	/**
 	 * @see AbstractMessageListenerContainer#setSubscriptionDurable(boolean)
 	 */
@@ -184,6 +194,9 @@ public abstract class AbstractJmsListenerContainerFactory<C extends AbstractMess
 		if (this.replyPubSubDomain != null) {
 			instance.setReplyPubSubDomain(this.replyPubSubDomain);
 		}
+		if (this.replyQosSettings != null) {
+			instance.setReplyQosSettings(this.replyQosSettings);
+		}
 		if (this.subscriptionDurable != null) {
 			instance.setSubscriptionDurable(this.subscriptionDurable);
 		}

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

@@ -28,6 +28,7 @@ import org.springframework.beans.factory.config.EmbeddedValueResolver;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.jms.listener.MessageListenerContainer;
 import org.springframework.jms.listener.adapter.MessagingMessageListenerAdapter;
+import org.springframework.jms.support.QosSettings;
 import org.springframework.jms.support.converter.MessageConverter;
 import org.springframework.jms.support.destination.DestinationResolver;
 import org.springframework.messaging.handler.annotation.SendTo;
@@ -147,6 +148,10 @@ public class MethodJmsListenerEndpoint extends AbstractJmsListenerEndpoint imple
 				messageListener.setDefaultResponseQueueName(responseDestination);
 			}
 		}
+		QosSettings responseQosSettings = container.getReplyQosSettings();
+		if (responseQosSettings != null) {
+			messageListener.setResponseQosSettings(responseQosSettings);
+		}
 		MessageConverter messageConverter = container.getMessageConverter();
 		if (messageConverter != null) {
 			messageListener.setMessageConverter(messageConverter);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -30,6 +30,7 @@ import javax.jms.Session;
 import javax.jms.Topic;
 
 import org.springframework.jms.support.JmsUtils;
+import org.springframework.jms.support.QosSettings;
 import org.springframework.jms.support.converter.MessageConverter;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -168,6 +169,8 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 
 	private Boolean replyPubSubDomain;
 
+	private QosSettings replyQosSettings;
+
 	private boolean pubSubNoLocal = false;
 
 	private MessageConverter messageConverter;
@@ -490,6 +493,22 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 		}
 	}
 
+	/**
+	 * Configure the {@link QosSettings} to use when sending a reply. Can be set to
+	 * {@code null} to indicate that the broker's defaults should be used.
+	 * @param replyQosSettings the QoS settings to use when sending a reply or {@code null}
+	 * to use the default vas.
+	 * @since 5.0
+	 */
+	public void setReplyQosSettings(QosSettings replyQosSettings) {
+		this.replyQosSettings = replyQosSettings;
+	}
+
+	@Override
+	public QosSettings getReplyQosSettings() {
+		return this.replyQosSettings;
+	}
+
 	/**
 	 * 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-2015 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -17,6 +17,7 @@
 package org.springframework.jms.listener;
 
 import org.springframework.context.SmartLifecycle;
+import org.springframework.jms.support.QosSettings;
 import org.springframework.jms.support.converter.MessageConverter;
 import org.springframework.jms.support.destination.DestinationResolver;
 
@@ -62,4 +63,11 @@ public interface MessageListenerContainer extends SmartLifecycle {
 	 */
 	boolean isReplyPubSubDomain();
 
+	/**
+	 * Return the {@link QosSettings} to use when sending a reply or {@code null}
+	 * if the broker's defaults should be used.
+	 * @since 5.0
+	 */
+	QosSettings getReplyQosSettings();
+
 }

spring-jms/src/main/java/org/springframework/jms/listener/adapter/AbstractAdaptableMessageListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -31,6 +31,7 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.jms.listener.SessionAwareMessageListener;
 import org.springframework.jms.support.JmsHeaderMapper;
 import org.springframework.jms.support.JmsUtils;
+import org.springframework.jms.support.QosSettings;
 import org.springframework.jms.support.SimpleJmsHeaderMapper;
 import org.springframework.jms.support.converter.MessageConversionException;
 import org.springframework.jms.support.converter.MessageConverter;
@@ -66,6 +67,7 @@ public abstract class AbstractAdaptableMessageListener
 
 	private final MessagingMessageConverterAdapter messagingMessageConverter = new MessagingMessageConverterAdapter();
 
+	private QosSettings responseQosSettings;
 
 	/**
 	 * Set the default destination to send response messages to. This will be applied
@@ -167,6 +169,24 @@ public abstract class AbstractAdaptableMessageListener
 		return this.messagingMessageConverter;
 	}
 
+	/**
+	 * Set the {@link QosSettings} to use when sending a response. Can be set to
+	 * {@code null} to indicate that the broker's defaults should be used.
+	 * @param responseQosSettings the QoS settings to use when sending a response or
+	 * {@code null} to use the default values.
+	 * @since 5.0
+	 */
+	public void setResponseQosSettings(QosSettings responseQosSettings) {
+		this.responseQosSettings = responseQosSettings;
+	}
+
+	/**
+	 * Return the {@link QosSettings} to use when sending a response or {@code null} if
+	 * the defaults should be used.
+	 */
+	protected QosSettings getResponseQosSettings() {
+		return this.responseQosSettings;
+	}
 
 	/**
 	 * Standard JMS {@link MessageListener} entry point.
@@ -399,7 +419,14 @@ public abstract class AbstractAdaptableMessageListener
 		MessageProducer producer = session.createProducer(destination);
 		try {
 			postProcessProducer(producer, response);
-			producer.send(response);
+			QosSettings settings = getResponseQosSettings();
+			if (settings != null) {
+				producer.send(response, settings.getDeliveryMode(), settings.getPriority(),
+						settings.getTimeToLive());
+			}
+			else {
+				producer.send(response);
+			}
 		}
 		finally {
 			JmsUtils.closeMessageProducer(producer);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -19,6 +19,7 @@ package org.springframework.jms.listener.endpoint;
 import javax.jms.Session;
 
 import org.springframework.core.Constants;
+import org.springframework.jms.support.QosSettings;
 import org.springframework.jms.support.converter.MessageConverter;
 
 /**
@@ -48,6 +49,8 @@ public class JmsActivationSpecConfig {
 
 	private Boolean replyPubSubDomain;
 
+	private QosSettings replyQosSettings;
+
 	private boolean subscriptionDurable = false;
 
 	private boolean subscriptionShared = false;
@@ -96,6 +99,14 @@ public class JmsActivationSpecConfig {
 		}
 	}
 
+	public void setReplyQosSettings(QosSettings replyQosSettings) {
+		this.replyQosSettings = replyQosSettings;
+	}
+
+	public QosSettings getReplyQosSettings() {
+		return this.replyQosSettings;
+	}
+
 	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-2015 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -22,6 +22,7 @@ import javax.resource.ResourceException;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.jca.endpoint.GenericMessageEndpointManager;
 import org.springframework.jms.listener.MessageListenerContainer;
+import org.springframework.jms.support.QosSettings;
 import org.springframework.jms.support.converter.MessageConverter;
 import org.springframework.jms.support.destination.DestinationResolver;
 
@@ -217,4 +218,12 @@ public class JmsMessageEndpointManager extends GenericMessageEndpointManager
 		throw new IllegalStateException("Could not determine reply pubSubDomain - no activation spec config is set");
 	}
 
+	@Override
+	public QosSettings getReplyQosSettings() {
+		JmsActivationSpecConfig config = getActivationSpecConfig();
+		if (config != null) {
+			return config.getReplyQosSettings();
+		}
+		throw new IllegalStateException("Could not determine reply qosSettings - no activation spec config is set");
+	}
 }

spring-jms/src/main/java/org/springframework/jms/support/QosSettings.java

@@ -0,0 +1,135 @@
+/*
+ * Copyright 2002-2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.jms.support;
+
+import javax.jms.Message;
+
+/**
+ * Gather the Quality of Service settings that can be used when sending a message.
+ *
+ * @author Stephane Nicoll
+ * @since 5.0
+ */
+public class QosSettings {
+
+	private int deliveryMode;
+
+	private int priority;
+
+	private long timeToLive;
+
+	/**
+	 * Create a new instance with the default settings.
+	 * @see Message#DEFAULT_DELIVERY_MODE
+	 * @see Message#DEFAULT_PRIORITY
+	 * @see Message#DEFAULT_TIME_TO_LIVE
+	 */
+	public QosSettings() {
+		this(Message.DEFAULT_DELIVERY_MODE, Message.DEFAULT_PRIORITY,
+				Message.DEFAULT_TIME_TO_LIVE);
+	}
+
+	/**
+	 * Create a new instance with the specified settings.
+	 */
+	public QosSettings(int deliveryMode, int priority, long timeToLive) {
+		this.deliveryMode = deliveryMode;
+		this.priority = priority;
+		this.timeToLive = timeToLive;
+	}
+
+	/**
+	 * Set the delivery mode to use when sending a message.
+	 * Default is the JMS Message default: "PERSISTENT".
+	 * @param deliveryMode the delivery mode to use
+	 * @see javax.jms.DeliveryMode#PERSISTENT
+	 * @see javax.jms.DeliveryMode#NON_PERSISTENT
+	 * @see javax.jms.Message#DEFAULT_DELIVERY_MODE
+	 * @see javax.jms.MessageProducer#send(javax.jms.Message, int, int, long)
+	 */
+	public void setDeliveryMode(int deliveryMode) {
+		this.deliveryMode = deliveryMode;
+	}
+
+	/**
+	 * Return the delivery mode to use when sending a message.
+	 */
+	public int getDeliveryMode() {
+		return this.deliveryMode;
+	}
+
+	/**
+	 * Set the priority of a message when sending.
+	 * @see javax.jms.Message#DEFAULT_PRIORITY
+	 * @see javax.jms.MessageProducer#send(javax.jms.Message, int, int, long)
+	 */
+	public void setPriority(int priority) {
+		this.priority = priority;
+	}
+
+	/**
+	 * Return the priority of a message when sending.
+	 */
+	public int getPriority() {
+		return this.priority;
+	}
+
+	/**
+	 * Set the time-to-live of the message when sending.
+	 * @param timeToLive the message's lifetime (in milliseconds)
+	 * @see javax.jms.Message#DEFAULT_TIME_TO_LIVE
+	 * @see javax.jms.MessageProducer#send(javax.jms.Message, int, int, long)
+	 */
+	public void setTimeToLive(long timeToLive) {
+		this.timeToLive = timeToLive;
+	}
+
+	/**
+	 * Return the time-to-live of the message when sending.
+	 */
+	public long getTimeToLive() {
+		return this.timeToLive;
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o) return true;
+		if (o == null || getClass() != o.getClass()) return false;
+
+		QosSettings that = (QosSettings) o;
+
+		if (this.deliveryMode != that.deliveryMode) return false;
+		if (this.priority != that.priority) return false;
+		return this.timeToLive == that.timeToLive;
+	}
+
+	@Override
+	public int hashCode() {
+		int result = this.deliveryMode;
+		result = 31 * result + this.priority;
+		result = 31 * result + (int) (this.timeToLive ^ (this.timeToLive >>> 32));
+		return result;
+	}
+
+	@Override
+	public String toString() {
+		return "QosSettings{" + "deliveryMode=" + deliveryMode +
+				", priority=" + priority +
+				", timeToLive=" + timeToLive +
+				'}';
+	}
+}