GH-6792: Add headers mapping to JMS channels (#10838)

* GH-6792: Add headers mapping to JMS channels

Fixes: #6792

Currently, by default `AbstractJmsChannel` implementations use Java serialization
for storing `Message<?>` instance into JMS destination.
The deserialized message is produced from the channel as is.
There might be use-cases when some JMS properties could be useful in downstream flows.

* Add `DefaultJmsHeaderMapper` into the `AbstractJmsChannel` as a conditional logic
to map headers to/from JMS when `jmsTemplate.getMessageConverter()` is not a `MessagingMessageConverter`.
Otherwise, then conversion of the message and header mapping is done in that converter
* Refactor logic in the `AbstractJmsChannel` to common API
* Fix Nullability formatting in the `PollableJmsChannel`
* Use lambda style for `MessageListener` in the `SubscribableJmsChannel`
instead of inner extra class
* Ensure in the `PollableJmsChannelTests` and `SubscribableJmsChannelTests`
that header mapper does it job
* Fix `JmsChannelParserTests` to not assert on the `messageBuilderFactory`
since now a `MessageListener` in the `SubscribableJmsChannel` is a direct method of the class
* Fix `JmsChannelHistoryTests` verification according to a new internal logic of the `AbstractJmsChannel`
* Document new feature in the `jms.adoc` and `whats-new.adoc`

# Conflicts:
#	src/reference/antora/modules/ROOT/pages/whats-new.adoc

* More fixes and cleanup for JMS Channels

* Use lambda-based `logger.debug()` in the `PollableJmsChannel` instead of `if (logger.isDebugEnabled()) {`:
the `message` variable is now effectively `final`
* Remove redundant null checks from the `SubscribableJmsChannel`
* Add more header tests into the `PollableJmsChannelTests` & `SubscribableJmsChannelTests`
* Clean up `JmsChannelParserTests` and verify the `messageBuilderFactory` settings

Author: artembilan

spring-integration-jms/src/main/java/org/springframework/integration/jms/channel/AbstractJmsChannel.java

@@ -16,18 +16,26 @@
 
 package org.springframework.integration.jms.channel;
 
+import jakarta.jms.JMSException;
+
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.jms.DefaultJmsHeaderMapper;
 import org.springframework.integration.jms.DynamicJmsTemplateProperties;
 import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.support.converter.MessageConverter;
+import org.springframework.jms.support.converter.MessagingMessageConverter;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 
 /**
  * A base {@link AbstractMessageChannel} implementation for JMS-backed message channels.
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 7.0
  *
@@ -36,27 +44,62 @@
  */
 public abstract class AbstractJmsChannel extends AbstractMessageChannel {
 
-	private final JmsTemplate jmsTemplate;
+	protected final DefaultJmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper();
+
+	protected final JmsTemplate jmsTemplate;
 
 	public AbstractJmsChannel(JmsTemplate jmsTemplate) {
 		Assert.notNull(jmsTemplate, "jmsTemplate must not be null");
 		this.jmsTemplate = jmsTemplate;
 	}
 
-	JmsTemplate getJmsTemplate() {
-		return this.jmsTemplate;
-	}
-
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
 		try {
 			DynamicJmsTemplateProperties.setPriority(new IntegrationMessageHeaderAccessor(message).getPriority());
-			this.jmsTemplate.convertAndSend(message);
+			MessageConverter messageConverter = this.jmsTemplate.getMessageConverter();
+			this.jmsTemplate.send((session) -> {
+				jakarta.jms.Message jmsMessage = messageConverter.toMessage(message, session);
+				if (!(messageConverter instanceof MessagingMessageConverter)) {
+					MessageHeaders headers = message.getHeaders();
+					this.headerMapper.fromHeaders(headers, jmsMessage);
+				}
+				return jmsMessage;
+			});
 		}
 		finally {
 			DynamicJmsTemplateProperties.clearPriority();
 		}
 		return true;
 	}
 
+	protected Message<?> fromJmsMessage(jakarta.jms.Message message) {
+		MessageConverter converter = this.jmsTemplate.getMessageConverter();
+		try {
+			Object converted = converter.fromMessage(message);
+			Message<?> messageToSend;
+			if (converted instanceof Message<?> convertedMessage) {
+				messageToSend = convertedMessage;
+				if (!(converter instanceof MessagingMessageConverter)) {
+					messageToSend = getMessageBuilderFactory()
+							.fromMessage(messageToSend)
+							.copyHeadersIfAbsent(this.headerMapper.toHeaders(message))
+							.build();
+				}
+			}
+			else {
+				messageToSend = getMessageBuilderFactory()
+						.withPayload(converted)
+						.copyHeaders(this.headerMapper.toHeaders(message))
+						.build();
+			}
+
+			return messageToSend;
+		}
+		catch (JMSException ex) {
+			throw new MessagingException("failed to convert incoming JMS Message", ex);
+		}
+
+	}
+
 }

spring-integration-jms/src/main/java/org/springframework/integration/jms/channel/PollableJmsChannel.java

@@ -61,8 +61,7 @@ public void setMessageSelector(String messageSelector) {
 	}
 
 	@Override
-	@Nullable
-	public Message<?> receive(long timeout) {
+	public @Nullable Message<?> receive(long timeout) {
 		try {
 			DynamicJmsTemplateProperties.setReceiveTimeout(timeout);
 			return receive();
@@ -73,8 +72,7 @@ public Message<?> receive(long timeout) {
 	}
 
 	@Override
-	@Nullable
-	public Message<?> receive() {
+	public @Nullable Message<?> receive() {
 		ChannelInterceptorList interceptorList = getIChannelInterceptorList();
 		Deque<ChannelInterceptor> interceptorStack = null;
 		boolean counted = false;
@@ -109,23 +107,22 @@ public Message<?> receive() {
 		}
 	}
 
-	@Nullable
-	private Message<?> receiveAndConvertToMessage() {
-		Object object;
+	private @Nullable Message<?> receiveAndConvertToMessage() {
+		jakarta.jms.Message jmsMessage;
 		if (this.messageSelector == null) {
-			object = getJmsTemplate().receiveAndConvert();
+			jmsMessage = this.jmsTemplate.receive();
 		}
 		else {
-			object = getJmsTemplate().receiveSelectedAndConvert(this.messageSelector);
+			jmsMessage = this.jmsTemplate.receiveSelected(this.messageSelector);
 		}
 
-		if (object == null) {
+		if (jmsMessage == null) {
 			logger.trace(() -> "postReceive on channel '" + this + "', message is null");
 			return null;
 		}
-		Message<?> message = object instanceof Message<?> msg
-				? msg
-				: getMessageBuilderFactory().withPayload(object).build();
+
+		Message<?> message = fromJmsMessage(jmsMessage);
+
 		logger.debug(() -> "postReceive on channel '" + this + "', message: " + message);
 
 		return message;

spring-integration-jms/src/main/java/org/springframework/integration/jms/channel/SubscribableJmsChannel.java

@@ -17,25 +17,19 @@
 package org.springframework.integration.jms.channel;
 
 import jakarta.jms.MessageListener;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 
 import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.channel.BroadcastCapableChannel;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
 import org.springframework.integration.dispatcher.BroadcastingDispatcher;
-import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
 import org.springframework.integration.dispatcher.UnicastingDispatcher;
-import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.integration.support.management.ManageableSmartLifecycle;
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
-import org.springframework.jms.support.converter.MessageConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandler;
-import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 
 /**
@@ -79,15 +73,11 @@ public void setMaxSubscribers(int maxSubscribers) {
 
 	@Override
 	public boolean subscribe(MessageHandler handler) {
-		Assert.state(this.dispatcher != null,
-				"'MessageDispatcher' must not be null. This channel might not have been initialized");
 		return this.dispatcher.addHandler(handler);
 	}
 
 	@Override
 	public boolean unsubscribe(MessageHandler handler) {
-		Assert.state(this.dispatcher != null,
-				"'MessageDispatcher' must not be null. This channel might not have been initialized");
 		return this.dispatcher.removeHandler(handler);
 	}
 
@@ -102,12 +92,8 @@ public void onInit() {
 			return;
 		}
 		super.onInit();
-		boolean isPubSub = isBroadcast();
-		configureDispatcher(isPubSub);
-		MessageListener listener =
-				new DispatchingMessageListener(getJmsTemplate(), this.dispatcher, this, isPubSub,
-						getMessageBuilderFactory());
-		this.container.setMessageListener(listener);
+		configureDispatcher(isBroadcast());
+		this.container.setMessageListener((MessageListener) this::receiveAndDispatch);
 		if (!this.container.isActive()) {
 			this.container.afterPropertiesSet();
 		}
@@ -140,115 +126,54 @@ private void configureDispatcher(boolean isPubSub) {
 
 	@Override
 	public boolean isAutoStartup() {
-		return (this.container != null) && this.container.isAutoStartup();
+		return this.container.isAutoStartup();
 	}
 
 	@Override
 	public int getPhase() {
-		return (this.container != null) ? this.container.getPhase() : 0;
+		return this.container.getPhase();
 	}
 
 	@Override
 	public boolean isRunning() {
-		return (this.container != null) && this.container.isRunning();
+		return this.container.isRunning();
 	}
 
 	@Override
 	public void start() {
-		if (this.container != null) {
-			this.container.start();
-		}
+		this.container.start();
 	}
 
 	@Override
 	public void stop() {
-		if (this.container != null) {
-			this.container.stop();
-		}
+		this.container.stop();
 	}
 
 	@Override
 	public void stop(Runnable callback) {
-		if (this.container != null) {
-			this.container.stop(callback);
-		}
-		else {
-			callback.run();
-		}
+		this.container.stop(callback);
 	}
 
 	@Override
 	public void destroy() {
-		if (this.container != null) {
-			this.container.destroy();
-		}
+		this.container.destroy();
 	}
 
-	private static final class DispatchingMessageListener implements MessageListener {
-
-		private final Log logger = LogFactory.getLog(this.getClass());
-
-		private final JmsTemplate jmsTemplate;
-
-		private final MessageDispatcher dispatcher;
-
-		private final SubscribableJmsChannel channel;
-
-		private final boolean isPubSub;
-
-		private final MessageBuilderFactory messageBuilderFactory;
-
-		DispatchingMessageListener(JmsTemplate jmsTemplate,
-				MessageDispatcher dispatcher, SubscribableJmsChannel channel, boolean isPubSub,
-				MessageBuilderFactory messageBuilderFactory) {
-
-			this.jmsTemplate = jmsTemplate;
-			this.dispatcher = dispatcher;
-			this.channel = channel;
-			this.isPubSub = isPubSub;
-			this.messageBuilderFactory = messageBuilderFactory;
+	private void receiveAndDispatch(jakarta.jms.Message message) {
+		Message<?> messageToSend = fromJmsMessage(message);
+		try {
+			this.dispatcher.dispatch(messageToSend);
 		}
-
-		@SuppressWarnings("NullAway") // Dataflow analysis limitation
-		@Override
-		public void onMessage(jakarta.jms.Message message) {
-			Message<?> messageToSend = null;
-			try {
-				MessageConverter converter = this.jmsTemplate.getMessageConverter();
-				Object converted = null;
-				if (converter != null) {
-					converted = converter.fromMessage(message);
-				}
-				if (converted != null) {
-					messageToSend =
-							converted instanceof Message<?> convertedMessage
-									? convertedMessage
-									: this.messageBuilderFactory.withPayload(converted).build();
-					this.dispatcher.dispatch(messageToSend);
-				}
-				else if (this.logger.isWarnEnabled()) {
-					this.logger.warn("No converter found, or converter returned null for: " + message
-							+ ", no Message to dispatch");
-				}
+		catch (MessageDispatchingException ex) {
+			String exceptionMessage = ex.getMessage() + " for jms-channel '" + this.getFullChannelName() + "'.";
+			if (isBroadcast()) {
+				// log only for backwards compatibility with pub/sub
+				this.logger.warn(ex, exceptionMessage);
 			}
-			catch (MessageDispatchingException ex) {
-				String exceptionMessage = ex.getMessage() + " for jms-channel '"
-						+ this.channel.getFullChannelName() + "'.";
-				if (this.isPubSub) {
-					// log only for backwards compatibility with pub/sub
-					if (this.logger.isWarnEnabled()) {
-						this.logger.warn(exceptionMessage, ex);
-					}
-				}
-				else {
-					throw new MessageDeliveryException(messageToSend, exceptionMessage, ex);
-				}
-			}
-			catch (Exception ex) {
-				throw new MessagingException("failed to handle incoming JMS Message", ex);
+			else {
+				throw new MessageDeliveryException(messageToSend, exceptionMessage, ex);
 			}
 		}
-
 	}
 
 }