INT-4317: JMS: dynamic deliverMode and timeToLive (#2561)

* INT-4317: JMS: dynamic deliverMode and timeToLive

JIRA: https://jira.spring.io/browse/INT-4317

* Add `deliveryModeExpression` and `timeToLiveExpression` properties
to the `JmsSendingMessageHandler` and expose them in the Java DSL and
XML components
* Add `setMapInboundDeliveryMode()` and `setMapInboundExpiration()`
`boolean` properties (default `false`) to the `DefaultJmsHeaderMapper`
for transferring `JMSDeliveryMode` and `JMSExpiration` into appropriate
`JmsHeaders.DELIVERY_MODE` and `JmsHeaders.EXPIRATION` headers
* Upgrade to latest Kotlin and AssertK

* Upgrade to Kotlin 1.2.61

Author: artembilan

build.gradle

@@ -1,5 +1,5 @@
 buildscript {
-	ext.kotlinVersion = '1.2.51'
+	ext.kotlinVersion = '1.2.61'
 	repositories {
 		maven { url 'https://repo.spring.io/plugins-release' }
 	}
@@ -88,7 +88,7 @@ subprojects { subproject ->
 		apacheSshdVersion = '1.7.0'
 		aspectjVersion = '1.9.0'
 		assertjVersion = '3.9.1'
-		assertkVersion = '0.10'
+		assertkVersion = '0.12'
 		boonVersion = '0.34'
 		commonsDbcp2Version = '2.2.0'
 		commonsIoVersion = '2.6'
@@ -178,9 +178,7 @@ subprojects { subproject ->
 		testRuntime "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion"
 		testRuntime "org.apache.logging.log4j:log4j-jcl:$log4jVersion"
 
-		testCompile("com.willowtreeapps.assertk:assertk:$assertkVersion") {
-				exclude group: 'org.jetbrains.kotlin', module: 'kotlin-reflect'
-		}
+		testCompile("com.willowtreeapps.assertk:assertk-jvm:$assertkVersion")
 
 		testCompile "org.jetbrains.kotlin:kotlin-stdlib-jdk8:$kotlinVersion"
 		testRuntime "org.jetbrains.kotlin:kotlin-reflect:$kotlinVersion"

spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -68,15 +68,35 @@ public class DefaultJmsHeaderMapper extends JmsHeaderMapper {
 
 	private volatile boolean mapInboundPriority = true;
 
+	private volatile boolean mapInboundDeliveryMode = false;
+
+	private volatile boolean mapInboundExpiration = false;
+
 	/**
 	 * Suppress the mapping of inbound priority by using this setter with 'false'.
-	 *
 	 * @param mapInboundPriority 'false' to suppress mapping the inbound priority.
 	 */
 	public void setMapInboundPriority(boolean mapInboundPriority) {
 		this.mapInboundPriority = mapInboundPriority;
 	}
 
+	/**
+	 * Map the inbound {@code deliveryMode} by using this setter with 'true'.
+	 * @param mapInboundDeliveryMode 'true' to map the inbound delivery mode.
+	 * @since 5.1
+	 */
+	public void setMapInboundDeliveryMode(boolean mapInboundDeliveryMode) {
+		this.mapInboundDeliveryMode = mapInboundDeliveryMode;
+	}
+	/**
+	 * Map the inbound {@code expiration} by using this setter with 'true'.
+	 * @param mapInboundExpiration 'true' to map the inbound expiration.
+	 * @since 5.1
+	 */
+	public void setMapInboundExpiration(boolean mapInboundExpiration) {
+		this.mapInboundExpiration = mapInboundExpiration;
+	}
+
 	/**
 	 * Specify a prefix to be appended to the integration message header name
 	 * for any JMS property that is being mapped into the MessageHeaders.
@@ -248,6 +268,22 @@ public Map<String, Object> toHeaders(javax.jms.Message jmsMessage) {
 					this.logger.info("failed to read JMSPriority property, skipping", e);
 				}
 			}
+			if (this.mapInboundDeliveryMode) {
+				try {
+					headers.put(JmsHeaders.DELIVERY_MODE, jmsMessage.getJMSDeliveryMode());
+				}
+				catch (Exception e) {
+					this.logger.info("failed to read JMSDeliveryMode property, skipping", e);
+				}
+			}
+			if (this.mapInboundExpiration) {
+				try {
+					headers.put(JmsHeaders.EXPIRATION, jmsMessage.getJMSExpiration());
+				}
+				catch (Exception e) {
+					this.logger.info("failed to read JMSExpiration property, skipping", e);
+				}
+			}
 			Enumeration<?> jmsPropertyNames = jmsMessage.getPropertyNames();
 			if (jmsPropertyNames != null) {
 				while (jmsPropertyNames.hasMoreElements()) {

spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplate.java

@@ -71,4 +71,16 @@ public long getReceiveTimeout() {
 		return (receiveTimeout != null) ? receiveTimeout : super.getReceiveTimeout();
 	}
 
+	@Override
+	public int getDeliveryMode() {
+		Integer deliveryMode = DynamicJmsTemplateProperties.getDeliveryMode();
+		return (deliveryMode != null) ? deliveryMode : super.getDeliveryMode();
+	}
+
+	@Override
+	public long getTimeToLive() {
+		Long timeToLive = DynamicJmsTemplateProperties.getTimeToLive();
+		return (timeToLive != null) ? timeToLive : super.getTimeToLive();
+	}
+
 }

spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplateProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -18,15 +18,24 @@
 
 /**
  * @author Mark Fisher
+ * @author Artem Bilan
+ *
  * @since 2.0.2
  */
 abstract class DynamicJmsTemplateProperties {
 
-	private static final ThreadLocal<Integer> priorityHolder = new ThreadLocal<Integer>();
+	private static final ThreadLocal<Integer> priorityHolder = new ThreadLocal<>();
+
+	private static final ThreadLocal<Long> receiveTimeoutHolder = new ThreadLocal<>();
+
+	private static final ThreadLocal<Integer> deliverModeHolder = new ThreadLocal<>();
 
-	private static final ThreadLocal<Long> receiveTimeoutHolder = new ThreadLocal<Long>();
+	private static final ThreadLocal<Long> timeToLiveHolder = new ThreadLocal<>();
 
 
+	private DynamicJmsTemplateProperties() {
+	}
+
 	public static Integer getPriority() {
 		return priorityHolder.get();
 	}
@@ -51,4 +60,28 @@ public static void clearReceiveTimeout() {
 		receiveTimeoutHolder.remove();
 	}
 
+	public static Integer getDeliveryMode() {
+		return deliverModeHolder.get();
+	}
+
+	public static void setDeliveryMode(Integer deliveryMode) {
+		deliverModeHolder.set(deliveryMode);
+	}
+
+	public static void clearDeliveryMode() {
+		deliverModeHolder.remove();
+	}
+
+	public static Long getTimeToLive() {
+		return timeToLiveHolder.get();
+	}
+
+	public static void setTimeToLive(Long timeToLive) {
+		timeToLiveHolder.set(timeToLive);
+	}
+
+	public static void clearTimeToLive() {
+		timeToLiveHolder.remove();
+	}
+
 }

spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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,11 +17,12 @@
 package org.springframework.integration.jms;
 
 import javax.jms.Destination;
-import javax.jms.JMSException;
 
 import org.springframework.core.convert.ConversionService;
+import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
-import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.StaticMessageHeaderAccessor;
+import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.jms.core.JmsTemplate;
@@ -35,20 +36,27 @@
  *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Artem Bilan
  */
 public class JmsSendingMessageHandler extends AbstractMessageHandler {
 
 	private final JmsTemplate jmsTemplate;
 
-	private volatile Destination destination;
+	private Destination destination;
 
-	private volatile String destinationName;
+	private String destinationName;
 
-	private volatile JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper();
+	private JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper();
 
-	private volatile boolean extractPayload = true;
+	private boolean extractPayload = true;
 
-	private volatile ExpressionEvaluatingMessageProcessor<?> destinationExpressionProcessor;
+	private ExpressionEvaluatingMessageProcessor<?> destinationExpressionProcessor;
+
+	private Expression deliveryModeExpression;
+
+	private Expression timeToLiveExpression;
+
+	private EvaluationContext evaluationContext;
 
 
 	public JmsSendingMessageHandler(JmsTemplate jmsTemplate) {
@@ -75,22 +83,61 @@ public void setDestinationExpression(Expression destinationExpression) {
 	}
 
 	public void setHeaderMapper(JmsHeaderMapper headerMapper) {
+		Assert.notNull(headerMapper, "'headerMapper' cannot be null");
 		this.headerMapper = headerMapper;
 	}
 
 	/**
 	 * Specify whether the payload should be extracted from each integration
 	 * Message to be used as the JMS Message body.
-	 *
 	 * <p>The default value is <code>true</code>. To force passing of the full
 	 * Spring Integration Message instead, set this to <code>false</code>.
-	 *
 	 * @param extractPayload true to extract the payload.
 	 */
 	public void setExtractPayload(boolean extractPayload) {
 		this.extractPayload = extractPayload;
 	}
 
+	/**
+	 * Specify a SpEL expression to evaluate a {@code deliveryMode} for the JMS message to send.
+	 * This option is applied only of QoS is enabled on the {@link JmsTemplate}.
+	 * @param deliveryModeExpression to use
+	 * @since 5.1
+	 * @see #setDeliveryModeExpression(Expression)
+	 */
+	public void setDeliveryModeExpressionString(String deliveryModeExpression) {
+		setDeliveryModeExpression(EXPRESSION_PARSER.parseExpression(deliveryModeExpression));
+	}
+
+	/**
+	 * Specify a SpEL expression to evaluate a {@code deliveryMode} for the JMS message to send.
+	 * This option is applied only of QoS is enabled on the {@link JmsTemplate}.
+	 * @param deliveryModeExpression to use
+	 * @since 5.1
+	 */
+	public void setDeliveryModeExpression(Expression deliveryModeExpression) {
+		this.deliveryModeExpression = deliveryModeExpression;
+	}
+
+	/**
+	 * Specify a SpEL expression to evaluate a {@code timeToLive} for the JMS message to send.
+	 * @param timeToLiveExpression to use
+	 * @since 5.1
+	 * @see #setTimeToLiveExpression(Expression)
+	 */
+	public void setTimeToLiveExpressionString(String timeToLiveExpression) {
+		setTimeToLiveExpression(EXPRESSION_PARSER.parseExpression(timeToLiveExpression));
+	}
+
+	/**
+	 * Specify a SpEL expression to evaluate a {@code timeToLive} for the JMS message to send.
+	 * @param timeToLiveExpression to use
+	 * @since 5.1
+	 */
+	public void setTimeToLiveExpression(Expression timeToLiveExpression) {
+		this.timeToLiveExpression = timeToLiveExpression;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "jms:outbound-channel-adapter";
@@ -105,22 +152,42 @@ protected void onInit() {
 				this.destinationExpressionProcessor.setConversionService(conversionService);
 			}
 		}
+		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
 	}
 
 	@Override
-	protected void handleMessageInternal(final Message<?> message) throws Exception {
-		if (message == null) {
-			throw new IllegalArgumentException("message must not be null");
-		}
+	protected void handleMessageInternal(final Message<?> message) {
 		Object destination = this.determineDestination(message);
 		Object objectToSend = (this.extractPayload) ? message.getPayload() : message;
 		MessagePostProcessor messagePostProcessor = new HeaderMappingMessagePostProcessor(message, this.headerMapper);
+
+		if (this.jmsTemplate instanceof DynamicJmsTemplate && this.jmsTemplate.isExplicitQosEnabled()) {
+			Integer priority = StaticMessageHeaderAccessor.getPriority(message);
+			if (priority != null) {
+				DynamicJmsTemplateProperties.setPriority(priority);
+			}
+			if (this.deliveryModeExpression != null) {
+				Integer deliveryMode =
+						this.deliveryModeExpression.getValue(this.evaluationContext, message, Integer.class);
+
+				if (deliveryMode != null) {
+					DynamicJmsTemplateProperties.setDeliveryMode(deliveryMode);
+				}
+			}
+			if (this.timeToLiveExpression != null) {
+				Long timeToLive = this.timeToLiveExpression.getValue(this.evaluationContext, message, Long.class);
+				if (timeToLive != null) {
+					DynamicJmsTemplateProperties.setTimeToLive(timeToLive);
+				}
+			}
+		}
 		try {
-			DynamicJmsTemplateProperties.setPriority(new IntegrationMessageHeaderAccessor(message).getPriority());
-			this.send(destination, objectToSend, messagePostProcessor);
+			send(destination, objectToSend, messagePostProcessor);
 		}
 		finally {
 			DynamicJmsTemplateProperties.clearPriority();
+			DynamicJmsTemplateProperties.clearDeliveryMode();
+			DynamicJmsTemplateProperties.clearTimeToLive();
 		}
 	}
 
@@ -167,10 +234,11 @@ private static final class HeaderMappingMessagePostProcessor implements MessageP
 		}
 
 		@Override
-		public javax.jms.Message postProcessMessage(javax.jms.Message jmsMessage) throws JMSException {
+		public javax.jms.Message postProcessMessage(javax.jms.Message jmsMessage) {
 			this.headerMapper.fromHeaders(this.integrationMessage.getHeaders(), jmsMessage);
 			return jmsMessage;
 		}
+
 	}
 
 }