diff --git a/build.gradle b/build.gradle index cf1148d8b3..da624a7318 100644 --- a/build.gradle +++ b/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" diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java index 755199b1e9..174b1a7798 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java +++ b/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 class DefaultJmsHeaderMapper extends JmsHeaderMapper { 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()) { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplate.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplate.java index 0d64fc8bff..52edf90228 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplate.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplate.java @@ -71,4 +71,16 @@ public class DynamicJmsTemplate extends JmsTemplate { 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(); + } + } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplateProperties.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplateProperties.java index c656047112..7d0b6f926b 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplateProperties.java +++ b/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,14 +18,23 @@ package org.springframework.integration.jms; /** * @author Mark Fisher + * @author Artem Bilan + * * @since 2.0.2 */ abstract class DynamicJmsTemplateProperties { - private static final ThreadLocal priorityHolder = new ThreadLocal(); + private static final ThreadLocal priorityHolder = new ThreadLocal<>(); - private static final ThreadLocal receiveTimeoutHolder = new ThreadLocal(); + private static final ThreadLocal receiveTimeoutHolder = new ThreadLocal<>(); + private static final ThreadLocal deliverModeHolder = new ThreadLocal<>(); + + private static final ThreadLocal timeToLiveHolder = new ThreadLocal<>(); + + + private DynamicJmsTemplateProperties() { + } public static Integer getPriority() { return priorityHolder.get(); @@ -51,4 +60,28 @@ abstract class DynamicJmsTemplateProperties { 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(); + } + } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java index 3be922e9a1..4dc3180702 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java +++ b/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 @@ import org.springframework.util.Assert; * * @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 class JmsSendingMessageHandler extends AbstractMessageHandler { } 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. - * *

The default value is true. To force passing of the full * Spring Integration Message instead, set this to false. - * * @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 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler { 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 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler { } @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; } + } } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java index 196404d04d..2cc194f166 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java @@ -57,8 +57,10 @@ public class JmsOutboundChannelAdapterParser extends AbstractOutboundChannelAdap if (hasDestinationRef || hasDestinationName || hasDestinationExpression) { if (!(hasDestinationRef ^ hasDestinationName ^ hasDestinationExpression)) { - parserContext.getReaderContext().error("The 'destination', 'destination-name', and " + - "'destination-expression' attributes are mutually exclusive.", parserContext.extractSource(element)); + parserContext.getReaderContext() + .error("The 'destination', 'destination-name', and 'destination-expression' " + + "attributes are mutually exclusive.", + parserContext.extractSource(element)); } if (hasDestinationRef) { builder.addPropertyReference(JmsParserUtils.DESTINATION_PROPERTY, destination); @@ -66,22 +68,30 @@ public class JmsOutboundChannelAdapterParser extends AbstractOutboundChannelAdap else if (hasDestinationName) { builder.addPropertyValue(JmsParserUtils.DESTINATION_NAME_PROPERTY, destinationName); } - else if (hasDestinationExpression) { - BeanDefinitionBuilder expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class); - expressionBuilder.addConstructorArgValue(destinationExpression); - builder.addPropertyValue(JmsParserUtils.DESTINATION_EXPRESSION_PROPERTY, expressionBuilder.getBeanDefinition()); + else { + BeanDefinitionBuilder expressionBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class) + .addConstructorArgValue(destinationExpression); + builder.addPropertyValue(JmsParserUtils.DESTINATION_EXPRESSION_PROPERTY, + expressionBuilder.getBeanDefinition()); } } else if (!hasJmsTemplate) { - parserContext.getReaderContext().error("either a '" + JmsParserUtils.JMS_TEMPLATE_ATTRIBUTE + - "' or one of '" + JmsParserUtils.DESTINATION_ATTRIBUTE + "', '" - + JmsParserUtils.DESTINATION_NAME_ATTRIBUTE + "', or '" + - JmsParserUtils.DESTINATION_EXPRESSION_ATTRIBUTE + - "' attributes must be provided", parserContext.extractSource(element)); + parserContext.getReaderContext() + .error("either a '" + JmsParserUtils.JMS_TEMPLATE_ATTRIBUTE + + "' or one of '" + JmsParserUtils.DESTINATION_ATTRIBUTE + "', '" + + JmsParserUtils.DESTINATION_NAME_ATTRIBUTE + "', or '" + + JmsParserUtils.DESTINATION_EXPRESSION_ATTRIBUTE + + "' attributes must be provided", parserContext.extractSource(element)); } - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, JmsParserUtils.HEADER_MAPPER_ATTRIBUTE); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + JmsParserUtils.HEADER_MAPPER_ATTRIBUTE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delivery-mode-expression", + "deliveryModeExpressionString"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "time-to-live-expression", + "timeToLiveExpressionString"); return builder.getBeanDefinition(); } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/Jms.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/Jms.java index f42d052a6e..a1b6d56d39 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/Jms.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/Jms.java @@ -193,6 +193,17 @@ public final class Jms { } } + /** + * The factory to produce a {@link JmsMessageDrivenChannelAdapterSpec}. + * @param jmsListenerContainerSpec the {@link JmsListenerContainerSpec} to build on + * @return the {@link JmsMessageDrivenChannelAdapterSpec} instance + */ + public static JmsMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( + JmsListenerContainerSpec jmsListenerContainerSpec) { + + return new JmsMessageDrivenChannelAdapterSpec<>(jmsListenerContainerSpec.get()); + } + /** * The factory to produce a {@link JmsMessageDrivenChannelAdapterSpec}. * @param listenerContainer the {@link AbstractMessageListenerContainer} to build on diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundChannelAdapterSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundChannelAdapterSpec.java index 368b6c631c..75d46a5083 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundChannelAdapterSpec.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundChannelAdapterSpec.java @@ -123,7 +123,53 @@ public class JmsOutboundChannelAdapterSpec S destination(Function, ?> destinationFunction) { - this.target.setDestinationExpression(new FunctionExpression>(destinationFunction)); + this.target.setDestinationExpression(new FunctionExpression<>(destinationFunction)); + return _this(); + } + + /** + * Specify a SpEL expression to evaluate a {@code deliveryMode} for JMS message to send. + * @param deliveryModeExpression to use + * @return the spec + * @since 5.1 + */ + public S deliveryModeExpression(String deliveryModeExpression) { + this.target.setDeliveryModeExpressionString(deliveryModeExpression); + return _this(); + } + + /** + * Specify a {@link Function} to resolve a {@code deliveryMode} for JMS message to send. + * @param deliveryModeFunction to use + * @return the spec + * @since 5.1 + * @see FunctionExpression + */ + public

S deliveryModeFunction(Function, ?> deliveryModeFunction) { + this.target.setDeliveryModeExpression(new FunctionExpression<>(deliveryModeFunction)); + return _this(); + } + + /** + * Specify a SpEL expression to evaluate a {@code timeToLive} for JMS message to send. + * @param timeToLiveExpression to use + * @return the spec + * @since 5.1 + */ + public S timeToLiveExpression(String timeToLiveExpression) { + this.target.setTimeToLiveExpressionString(timeToLiveExpression); + return _this(); + } + + /** + * Specify a {@link Function} to resolve a {@code timeToLive} for JMS message to send. + * @param timeToLiveFunction to use + * @return the spec + * @see FunctionExpression + * @since 5.1 + */ + public

S timeToLiveFunction(Function, ?> timeToLiveFunction) { + this.target.setTimeToLiveExpression(new FunctionExpression<>(timeToLiveFunction)); return _this(); } diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-5.1.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-5.1.xsd index c1097901c4..c194bc6dcc 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-5.1.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-5.1.xsd @@ -1259,6 +1259,28 @@ ]]> + + + + A SpEL expression to be evaluated at runtime against each Spring Integration Message as + the root object. The result should be an integer representing the delivery mode. + If returns null, falls back to the 'delivery-persistent' or default. + The explicit-qos-enabled has to be enabled. + Note: the static delivery mode value can be specified on the JmsTemplate. + + + + + + + A SpEL expression to be evaluated at runtime against each Spring Integration Message as + the root object. The result should be a long representing the time-to-live for the message. + If returns null, falls back to the 'time-to-live' or default. + The 'explicit-qos-enabled' has to be enabled. + Note: the static time-to-live value can be specified on the JmsTemplate. + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParserTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParserTests.java index 33578bc0e5..1abc44da91 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParserTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParserTests.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. @@ -41,6 +41,7 @@ import org.springframework.messaging.support.GenericMessage; /** * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ public class JmsOutboundChannelAdapterParserTests { @@ -134,13 +135,16 @@ public class JmsOutboundChannelAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsOutboundWithJmsTemplateQos.xml", this.getClass()); EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("adapter"); - DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(new DirectFieldAccessor(endpoint).getPropertyValue("handler")); + Object handler = new DirectFieldAccessor(endpoint).getPropertyValue("handler"); + DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); JmsTemplate jmsTemplate = (JmsTemplate) handlerAccessor.getPropertyValue("jmsTemplate"); assertNotNull(jmsTemplate); assertEquals(context.getBean("template"), jmsTemplate); assertTrue(jmsTemplate.isExplicitQosEnabled()); assertEquals(7, jmsTemplate.getPriority()); assertEquals(12345, jmsTemplate.getTimeToLive()); + assertEquals("1", TestUtils.getPropertyValue(handler, "deliveryModeExpression.expression", String.class)); + assertEquals("100", TestUtils.getPropertyValue(handler, "timeToLiveExpression.expression", String.class)); context.close(); } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundWithJmsTemplateQos.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundWithJmsTemplateQos.xml index 77bb36941f..86099e4295 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundWithJmsTemplateQos.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundWithJmsTemplateQos.xml @@ -1,9 +1,9 @@ - + - + diff --git a/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt b/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt index e6c099e61d..80e0f3b85a 100644 --- a/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt +++ b/spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt @@ -16,7 +16,10 @@ package org.springframework.integration.jms.dsl +import assertk.assert +import assertk.assertions.contains import assertk.assertions.isEqualTo +import assertk.assertions.isGreaterThan import assertk.assertions.isNotNull import org.apache.activemq.ActiveMQConnectionFactory import org.junit.jupiter.api.Test @@ -24,17 +27,21 @@ import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Qualifier import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration +import org.springframework.integration.IntegrationMessageHeaderAccessor import org.springframework.integration.config.EnableIntegration import org.springframework.integration.dsl.IntegrationFlow import org.springframework.integration.dsl.IntegrationFlows import org.springframework.integration.dsl.MessageChannels +import org.springframework.integration.jms.DefaultJmsHeaderMapper import org.springframework.integration.support.MessageBuilder +import org.springframework.jms.support.JmsHeaders import org.springframework.messaging.MessageChannel import org.springframework.messaging.PollableChannel import org.springframework.messaging.simp.SimpMessageHeaderAccessor import org.springframework.test.annotation.DirtiesContext import org.springframework.test.context.junit.jupiter.SpringJUnitConfig import java.util.concurrent.Executors +import javax.jms.DeliveryMode /** * @author Artem Bilan @@ -57,15 +64,24 @@ class JmsDslKotlinTests { this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload(" foo ") .setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "containerSpecDestination") + .setPriority(9) .build()) val receive = this.jmsOutboundInboundReplyChannel.receive(10000) val payload = receive?.payload - assertk.assert(payload).isNotNull { + assert(payload).isNotNull { it.isEqualTo("foo") } + + assert(receive?.headers).isNotNull { + it.contains(IntegrationMessageHeaderAccessor.PRIORITY, 9) + it.contains(JmsHeaders.DELIVERY_MODE, 1) + } + + val expiration = receive!!.headers[JmsHeaders.EXPIRATION] as Long + assert(expiration).isGreaterThan(System.currentTimeMillis()) } @Configuration @@ -83,24 +99,35 @@ class JmsDslKotlinTests { fun jmsOutboundFlow() = IntegrationFlow { f -> f.handle(Jms.outboundAdapter(jmsConnectionFactory()) - .destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)) + .destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER) + .deliveryModeFunction { _ -> DeliveryMode.NON_PERSISTENT } + .timeToLiveExpression("10000") + .configureJmsTemplate { t -> t.explicitQosEnabled(true) }) } + @Bean + fun jmsHeaderMapper(): DefaultJmsHeaderMapper { + val jmsHeaderMapper = DefaultJmsHeaderMapper() + jmsHeaderMapper.setMapInboundDeliveryMode(true) + jmsHeaderMapper.setMapInboundExpiration(true) + return jmsHeaderMapper + } + + @Bean + fun jmsOutboundInboundReplyChannel() = MessageChannels.queue().get() + @Bean fun jmsMessageDrivenFlowWithContainer() = IntegrationFlows.from( Jms.messageDrivenChannelAdapter( Jms.container(jmsConnectionFactory(), "containerSpecDestination") .pubSubDomain(false) - .taskExecutor(Executors.newCachedThreadPool()) - .get())) + .taskExecutor(Executors.newCachedThreadPool())) + .headerMapper(jmsHeaderMapper())) .transform({ it: String -> it.trim({ it <= ' ' }) }) .channel(jmsOutboundInboundReplyChannel()) .get() - @Bean - fun jmsOutboundInboundReplyChannel() = MessageChannels.queue().get() - } diff --git a/src/reference/asciidoc/jms.adoc b/src/reference/asciidoc/jms.adoc index 374a8a229d..6ff44000b2 100644 --- a/src/reference/asciidoc/jms.adoc +++ b/src/reference/asciidoc/jms.adoc @@ -171,6 +171,17 @@ If, on the other hand, you want to send the actual Spring Integration message to NOTE: Regardless of the boolean value for payload extraction, the Spring Integration `MessageHeaders` map to JMS properties, as long as you rely on the default converter or provide a reference to another instance of `HeaderMappingMessageConverter`. (The same holds true for 'inbound' adapters, except that, in those cases, the JMS properties map to Spring Integration `MessageHeaders`). +Starting with version 5.1, the `` (`JmsSendingMessageHandler`) can be configured with the `deliveryModeExpression` and `timeToLiveExpression` properties to evaluate an appropriate QoS values for JMS message to send at runtime against request Spring `Message`. +The new `setMapInboundDeliveryMode(true)` and `setMapInboundExpiration(true)` options of the `DefaultJmsHeaderMapper` may facilitate as a source of the information for the dynamic `deliveryMode` and `timeToLive` from message headers: + +==== +[source,xml] +---- + +---- +==== + [[jms-ob-transactions]] ==== Transactions @@ -585,6 +596,22 @@ On the inbound side, it is mapped as a `String`. This is independent of the `jms_correlationId` header, which is mapped to and from the `JMSCorrelationID` header. The `JMSCorrelationID` is generally used to correlate requests and replies, whereas the `correlationId` is often used to combine related messages into a group (such as with an aggregator or a resequencer). +Starting with version 5.1, the `DefaultJmsHeaderMapper` can be configured for mapping inbound `JMSDeliveryMode` and `JMSExpiration` properties: + ==== + [source,java] + ---- + @Bean + public DefaultJmsHeaderMapper jmsHeaderMapper() { + DefaultJmsHeaderMapper mapper = new DefaultJmsHeaderMapper(); + mapper.setMapInboundDeliveryMode(true) + mapper.setMapInboundExpiration(true) + return mapper; + } + ---- + ==== + +These JMS properties are mapped to the `JmsHeaders.DELIVERY_MODE` and `JmsHeaders.EXPIRATION` Spring Message headers respectively. + [[jms-conversion-and-marshalling]] === Message Conversion, Marshalling, and Unmarshalling @@ -593,6 +620,7 @@ To do so, provide the bean name of an instance of `MessageConverter` that is ava Also, to provide some consistency with marshaller and unmarshaller interfaces, Spring provides `MarshallingMessageConverter`, which you can configure with your own custom marshallers and unmarshallers. The following example shows how to do so +==== [source,xml] ---- ---- +==== NOTE: When you provide your own `MessageConverter` instance, it is still wrapped within the `HeaderMappingMessageConverter`. This means that the 'extract-request-payload' and 'extract-reply-payload' properties can affect the actual objects passed to your converter. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 2313072114..82932bc9c7 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -152,3 +152,11 @@ In addition the key and trust store types can now be configured on the `DefaultT Since the Spring Social project has moved to https://spring.io/blog/2018/07/03/spring-social-end-of-life-announcement[end of life status], Twitter support in Spring Integration has been moved to the Extensions project. See https://github.com/spring-projects/spring-integration-extensions/tree/master/spring-integration-social-twitter[Spring Integration Social Twitter] for more information. + +[[x51.-jms]] +=== JMS Support + +The `JmsSendingMessageHandler` now provides `deliveryModeExpression` and `timeToLiveExpression` options to determine respective QoS options for JMS message to send at runtime. +The `DefaultJmsHeaderMapper` now allows to map inbound `JMSDeliveryMode` and `JMSExpiration` properties via setting to `true` respective `setMapInboundDeliveryMode()` and `setMapInboundExpiration()` options. + +See <> for more information.