INT-3784: Use acknowledge="transacted" by Default

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

Also

- suppress WARN log when priority mapping fails
- rework extract payload tests to use a single context

Make a new `JmsMessageDrivenEndpoint` as `private` because it makes sense only for XML `BeanDefinition` variant.

Author: garyrussell

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

@@ -341,6 +341,9 @@ public void onMessage(javax.jms.Message jmsMessage, Session session) throws JMSE
 				Message<?> replyMessage = this.gatewayDelegate.sendAndReceiveMessage(requestMessage);
 				if (replyMessage != null) {
 					Destination destination = this.getReplyDestination(jmsMessage, session);
+					if (logger.isDebugEnabled()) {
+						logger.debug("Reply destination: " + destination);
+					}
 					if (destination != null) {
 						// convert SI Message to JMS Message
 						Object replyResult = replyMessage;

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

@@ -150,7 +150,8 @@ public void fromHeaders(MessageHeaders headers, javax.jms.Message jmsMessage) {
 							jmsMessage.setObjectProperty(propertyName, value);
 						}
 						catch (Exception e) {
-							if (headerName.startsWith("JMSX")) {
+							if (headerName.startsWith("JMSX")
+									|| headerName.equals(IntegrationMessageHeaderAccessor.PRIORITY)) {
 								if (logger.isTraceEnabled()) {
 									logger.trace("skipping reserved header, it cannot be set by client: " + headerName);
 								}

spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.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.
@@ -21,6 +21,7 @@
 import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.jms.util.JmsAdapterUtils;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
 import org.springframework.util.Assert;
 
 /**
@@ -35,12 +36,33 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos
 
 	private final AbstractMessageListenerContainer listenerContainer;
 
+	private final boolean externalContainer;
+
 	private final ChannelPublishingJmsMessageListener listener;
-	
+
 	private volatile String sessionAcknowledgeMode;
-	
+
+	/**
+	 * Construct an instance with an externally configured container.
+	 * @param listenerContainer the container.
+	 * @param listener the listener.
+	 */
 	public JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer,
 			ChannelPublishingJmsMessageListener listener) {
+		this(listenerContainer, listener, true);
+	}
+
+	/**
+	 * Construct an instance with an argument indicating whether the container's ack mode should
+	 * be overridden with {@link #setSessionAcknowledgeMode(String) sessionAcknowledgeMode}, default
+	 * 'transacted'.
+	 * @param listenerContainer the container.
+	 * @param listener the listener.
+	 * @param externalContainer true if the container is externally configured and should not have its ackmode
+	 * coerced when no sessionAcknowledgeMode was supplied.
+	 */
+	private JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer,
+	                                 ChannelPublishingJmsMessageListener listener, boolean externalContainer) {
 		Assert.notNull(listenerContainer, "listener container must not be null");
 		Assert.notNull(listener, "listener must not be null");
 		if (logger.isWarnEnabled() && listenerContainer.getMessageListener() != null) {
@@ -51,12 +73,20 @@ public JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContain
 		this.listener = listener;
 		this.listenerContainer = listenerContainer;
 		setPhase(Integer.MAX_VALUE / 2);
+		this.externalContainer = externalContainer;
 	}
 
+	/**
+	 * Set the session acknowledge mode on the listener container. It will override the
+	 * container setting even if an external container is provided. Defaults to null
+	 * (won't change container) if an external container is provided or `transacted` when
+	 * the framework creates an implicit {@link DefaultMessageListenerContainer}.
+	 * @param sessionAcknowledgeMode the acknowledge mode.
+	 */
 	public void setSessionAcknowledgeMode(String sessionAcknowledgeMode) {
 		this.sessionAcknowledgeMode = sessionAcknowledgeMode;
 	}
-	
+
 	@Override
 	public String getComponentType() {
 		return "jms:message-driven-channel-adapter";
@@ -68,7 +98,12 @@ protected void onInit() throws Exception {
 		if (!this.listenerContainer.isActive()) {
 			this.listenerContainer.afterPropertiesSet();
 		}
-		Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(this.sessionAcknowledgeMode);
+		String sessionAcknowledgeMode = this.sessionAcknowledgeMode;
+		if (sessionAcknowledgeMode == null && !this.externalContainer
+				&& DefaultMessageListenerContainer.class.isAssignableFrom(this.listenerContainer.getClass())) {
+			sessionAcknowledgeMode = JmsAdapterUtils.SESSION_TRANSACTED_STRING;
+		}
+		Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(sessionAcknowledgeMode);
 		if (acknowledgeMode != null) {
 			if (acknowledgeMode.intValue() == JmsAdapterUtils.SESSION_TRANSACTED) {
 				this.listenerContainer.setSessionTransacted(true);
@@ -94,6 +129,7 @@ protected void doStop() {
 		this.listener.stop();
 	}
 
+	@Override
 	public void destroy() throws Exception {
 		if (this.isRunning()) {
 			this.stop();
@@ -102,12 +138,14 @@ public void destroy() throws Exception {
 	}
 
 
+	@Override
 	public int beforeShutdown() {
 		this.stop();
 		return 0;
 	}
 
 
+	@Override
 	public int afterShutdown() {
 		return 0;
 	}

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

@@ -770,6 +770,9 @@ private javax.jms.Message sendAndReceiveWithContainer(Message<?> requestMessage)
 
 			jmsRequest.setJMSReplyTo(replyTo);
 			connection.start();
+			if (logger.isDebugEnabled()) {
+				logger.debug("ReplyTo: " + replyTo);
+			}
 
 			Integer priority = new IntegrationMessageHeaderAccessor(requestMessage).getPriority();
 			if (priority == null) {
@@ -824,6 +827,9 @@ private javax.jms.Message sendAndReceiveWithoutContainer(Message<?> requestMessa
 			replyTo = this.determineReplyDestination(requestMessage, session);
 			jmsRequest.setJMSReplyTo(replyTo);
 			connection.start();
+			if (logger.isDebugEnabled()) {
+				logger.debug("ReplyTo: " + replyTo);
+			}
 
 			Integer priority = new IntegrationMessageHeaderAccessor(requestMessage).getPriority();
 			if (priority == null) {

spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java

@@ -63,6 +63,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
 		JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE,
 		"destination-resolver", "transaction-manager",
 		"concurrent-consumers", "max-concurrent-consumers",
+		"acknowledge",
 		"max-messages-per-task", "selector",
 		"receive-timeout", "recovery-interval",
 		"idle-consumer-limit", "idle-task-execution-limit",
@@ -117,6 +118,7 @@ protected void doParse(Element element, ParserContext parserContext, BeanDefinit
 		String listenerBeanName = this.parseMessageListener(element, parserContext, builder.getRawBeanDefinition());
 		builder.addConstructorArgReference(containerBeanName);
 		builder.addConstructorArgReference(listenerBeanName);
+		builder.addConstructorArgValue(hasExternalContainer(element));
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge", "sessionAcknowledgeMode");
@@ -126,7 +128,7 @@ protected void doParse(Element element, ParserContext parserContext, BeanDefinit
 	private String parseMessageListenerContainer(Element element, ParserContext parserContext,
 			BeanDefinition adapterBeanDefinition) {
 		String containerClass = element.getAttribute("container-class");
-		if (element.hasAttribute("container")) {
+		if (hasExternalContainer(element)) {
 			if (StringUtils.hasText(containerClass)) {
 				parserContext.getReaderContext().error("Cannot have both 'container' and 'container-class'", element);
 			}
@@ -198,6 +200,11 @@ private String parseMessageListenerContainer(Element element, ParserContext pars
 		return beanName;
 	}
 
+
+	private boolean hasExternalContainer(Element element) {
+		return element.hasAttribute("container");
+	}
+
 	private String parseMessageListener(Element element, ParserContext parserContext,
 			BeanDefinition adapterBeanDefinition) {
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder

spring-integration-jms/src/main/java/org/springframework/integration/jms/util/JmsAdapterUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-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.
@@ -19,11 +19,20 @@
 
 /**
  * @author Liujiong
+ * @author Gary Russell
  * @since 4.1
  *
  */
 public abstract class JmsAdapterUtils {
-	
+
+	public static final String AUTO_ACKNOWLEDGE_STRING = "auto";
+
+	public static final String DUPS_OK_ACKNOWLEDGE_STRING = "dups-ok";
+
+	public static final String CLIENT_ACKNOWLEDGE_STRING = "client";
+
+	public static final String SESSION_TRANSACTED_STRING = "transacted";
+
 	public static final int SESSION_TRANSACTED = 0;
 
 	public static final int AUTO_ACKNOWLEDGE = 1;
@@ -35,16 +44,16 @@ public abstract class JmsAdapterUtils {
 	public static Integer parseAcknowledgeMode(String acknowledge) {
 		if (StringUtils.hasText(acknowledge)) {
 			int acknowledgeMode = AUTO_ACKNOWLEDGE;
-			if ("transacted".equals(acknowledge)) {
+			if (SESSION_TRANSACTED_STRING.equals(acknowledge)) {
 				acknowledgeMode = SESSION_TRANSACTED;
 			}
-			else if ("dups-ok".equals(acknowledge)) {
+			else if (DUPS_OK_ACKNOWLEDGE_STRING.equals(acknowledge)) {
 				acknowledgeMode = DUPS_OK_ACKNOWLEDGE;
 			}
-			else if ("client".equals(acknowledge)) {
+			else if (CLIENT_ACKNOWLEDGE_STRING.equals(acknowledge)) {
 				acknowledgeMode = CLIENT_ACKNOWLEDGE;
 			}
-			else if (!"auto".equals(acknowledge)) {
+			else if (!AUTO_ACKNOWLEDGE_STRING.equals(acknowledge)) {
 				throw new IllegalStateException("Invalid JMS 'acknowledge' setting: " +
 						"only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported.");
 			}

spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd

@@ -1317,7 +1317,10 @@
 	The native JMS acknowledge mode: "auto", "client", "dups-ok" or "transacted".
 	The latter effectively activates a locally transacted Session. 'transacted'
 	is not allowed on the inbound-channel-adapter; use 'session-transacted' instead.
-	acknowlege="transacted" is used on the message-driven-channel-adapter.
+	acknowlege="transacted" is used on the message-driven-channel-adapter and inbound gateway.
+	Defaults to "transacted" when an implicit message listener container is configured.
+	Not allowed when using an externally configured listener container; configure the container
+	instead.
 						]]></xsd:documentation>
 					</xsd:annotation>
 					<xsd:simpleType>