GH-7925: Make message history header as mutable

spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -23,8 +23,11 @@
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.integration.core.MessageSelector;
+import org.springframework.integration.history.MessageHistory;
+import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.integration.support.channel.ChannelResolverUtils;
 import org.springframework.integration.support.management.ManageableLifecycle;
+import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.jmx.export.annotation.ManagedAttribute;
 import org.springframework.jmx.export.annotation.ManagedOperation;
 import org.springframework.jmx.export.annotation.ManagedResource;
@@ -57,6 +60,8 @@ public class WireTap implements ChannelInterceptor, ManageableLifecycle, VetoCap
 
 	private BeanFactory beanFactory;
 
+	private MessageBuilderFactory messageBuilderFactory;
+
 	private volatile boolean running = true;
 
 
@@ -162,9 +167,19 @@ public Message<?> preSend(Message<?> message, MessageChannel channel) {
 			return message;
 		}
 		if (this.running && (this.selector == null || this.selector.accept(message))) {
-			boolean sent = (this.timeout >= 0)
-					? wireTapChannel.send(message, this.timeout)
-					: wireTapChannel.send(message);
+			Message<?> messageToSend = message;
+			if (message.getHeaders().containsKey(MessageHistory.HEADER_NAME)) {
+				messageToSend =
+						getMessageBuilderFactory()
+								.fromMessage(message)
+								.cloneMessageHistoryIfAny()
+								.build();
+			}
+			boolean sent =
+					(this.timeout >= 0)
+							? wireTapChannel.send(messageToSend, this.timeout)
+							: wireTapChannel.send(messageToSend);
+
 			if (!sent && LOGGER.isWarnEnabled()) {
 				LOGGER.warn("failed to send message to WireTap channel '" + wireTapChannel + "'");
 			}
@@ -174,7 +189,6 @@ public Message<?> preSend(Message<?> message, MessageChannel channel) {
 
 	@Override
 	public boolean shouldIntercept(String beanName, InterceptableChannel channel) {
-
 		return !getChannel().equals(channel);
 	}
 
@@ -189,4 +203,11 @@ private MessageChannel getChannel() {
 		return this.channel;
 	}
 
+	private MessageBuilderFactory getMessageBuilderFactory() {
+		if (this.messageBuilderFactory == null) {
+			this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
+		}
+		return this.messageBuilderFactory;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2024 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,18 +17,23 @@
 package org.springframework.integration.dispatcher;
 
 import java.util.Collection;
+import java.util.UUID;
 import java.util.concurrent.Executor;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.integration.MessageDispatchingException;
+import org.springframework.integration.context.IntegrationObjectSupport;
+import org.springframework.integration.history.MessageHistory;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.DefaultMessageBuilderFactory;
 import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.integration.support.MessageDecorator;
 import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.MessageHandlingRunnable;
 import org.springframework.util.Assert;
@@ -153,13 +158,28 @@ public boolean dispatch(Message<?> message) {
 		}
 		int sequenceSize = handlers.size();
 		Message<?> messageToSend = message;
+		MessageHeaders messageHeaders = message.getHeaders();
+		UUID correlationKey = messageHeaders.getId();
+		boolean hasMessageHistory = messageHeaders.containsKey(MessageHistory.HEADER_NAME) && sequenceSize > 1;
 		for (MessageHandler handler : handlers) {
-			if (this.applySequence) {
-				messageToSend = getMessageBuilderFactory()
-						.fromMessage(message)
-						.pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize)
-						.build();
+			if (this.applySequence || hasMessageHistory) {
+				AbstractIntegrationMessageBuilder<?> builder =
+						getMessageBuilderFactory()
+								.fromMessage(message);
+
+				if (this.applySequence) {
+					builder.pushSequenceDetails(
+							correlationKey == null ? IntegrationObjectSupport.generateId() : correlationKey,
+							sequenceNumber++, sequenceSize);
+				}
+
+				if (hasMessageHistory) {
+					builder.cloneMessageHistoryIfAny();
+				}
+
+				messageToSend = builder.build();
 			}
+
 			if (message instanceof MessageDecorator messageDecorator) {
 				messageToSend = messageDecorator.decorateMessage(messageToSend);
 			}

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java

@@ -20,6 +20,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -465,6 +466,13 @@ else if (output instanceof AbstractIntegrationMessageBuilder) {
 		}
 		else {
 			builder = getMessageBuilderFactory().withPayload(output);
+			// Assuming that message in the payload collection is a copy of request message.
+			if (output instanceof Iterable<?> iterable) {
+				Iterator<?> iterator = iterable.iterator();
+				if (iterator.hasNext() && iterator.next() instanceof Message<?>) {
+					builder = builder.cloneMessageHistoryIfAny();
+				}
+			}
 		}
 		if (!this.noHeadersPropagation &&
 				(shouldCopyRequestHeaders() ||

spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.history;
 
+import java.io.Serial;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -52,8 +53,9 @@
  *
  * @since 2.0
  */
-public final class MessageHistory implements List<Properties>, Serializable {
+public final class MessageHistory implements List<Properties>, Serializable, Cloneable {
 
+	@Serial
 	private static final long serialVersionUID = -2340400235574314134L;
 
 	private static final Log LOGGER = LogFactory.getLog(MessageHistory.class);
@@ -94,51 +96,52 @@ public static <T> Message<T> write(Message<T> messageArg, NamedComponent compone
 		Assert.notNull(component, "Component must not be null");
 		Properties metadata = extractMetadata(component);
 		if (!metadata.isEmpty()) {
-			MessageHistory previousHistory = message.getHeaders().get(HEADER_NAME, MessageHistory.class);
-			List<Properties> components =
-					previousHistory != null
-							? new ArrayList<>(previousHistory)
-							: new ArrayList<>();
-			components.add(metadata);
-			MessageHistory history = new MessageHistory(components);
-
-			if (message instanceof MutableMessage) {
-				message.getHeaders().put(HEADER_NAME, history);
+			MessageHistory messageHistory = message.getHeaders().get(HEADER_NAME, MessageHistory.class);
+			if (messageHistory != null) {
+				messageHistory.components.add(metadata);
 			}
-			else if (message instanceof ErrorMessage) {
-				ErrorMessage errorMessage = (ErrorMessage) message;
-				IntegrationMessageHeaderAccessor headerAccessor = new IntegrationMessageHeaderAccessor(message);
-				headerAccessor.setHeader(HEADER_NAME, history);
-				Throwable payload = errorMessage.getPayload();
-				Message<?> originalMessage = errorMessage.getOriginalMessage();
-				if (originalMessage != null) {
-					errorMessage = new ErrorMessage(payload, headerAccessor.toMessageHeaders(), originalMessage);
+			else {
+				List<Properties> components = new ArrayList<>();
+				components.add(metadata);
+				messageHistory = new MessageHistory(components);
+
+				if (message instanceof MutableMessage) {
+					message.getHeaders().put(HEADER_NAME, messageHistory);
 				}
-				else {
-					errorMessage = new ErrorMessage(payload, headerAccessor.toMessageHeaders());
+				else if (message instanceof ErrorMessage errorMessage) {
+					IntegrationMessageHeaderAccessor headerAccessor = new IntegrationMessageHeaderAccessor(message);
+					headerAccessor.setHeader(HEADER_NAME, messageHistory);
+					Throwable payload = errorMessage.getPayload();
+					Message<?> originalMessage = errorMessage.getOriginalMessage();
+					if (originalMessage != null) {
+						errorMessage = new ErrorMessage(payload, headerAccessor.toMessageHeaders(), originalMessage);
+					}
+					else {
+						errorMessage = new ErrorMessage(payload, headerAccessor.toMessageHeaders());
+					}
+					message = (Message<T>) errorMessage;
 				}
-				message = (Message<T>) errorMessage;
-			}
-			else if (message instanceof AdviceMessage) {
-				IntegrationMessageHeaderAccessor headerAccessor = new IntegrationMessageHeaderAccessor(message);
-				headerAccessor.setHeader(HEADER_NAME, history);
-				message = new AdviceMessage<T>(message.getPayload(), headerAccessor.toMessageHeaders(),
-						((AdviceMessage<?>) message).getInputMessage());
-			}
-			else {
-				if (!(message instanceof GenericMessage) &&
-						(messageBuilderFactory instanceof DefaultMessageBuilderFactory ||
-								messageBuilderFactory instanceof MutableMessageBuilderFactory)
-						&& LOGGER.isWarnEnabled()) {
-
-					LOGGER.warn("MessageHistory rebuilds the message and produces the result of the [" +
-							messageBuilderFactory + "], not an instance of the provided type [" +
-							message.getClass() + "]. Consider to supply a custom MessageBuilderFactory " +
-							"to retain custom messages during MessageHistory tracking.");
+				else if (message instanceof AdviceMessage<?> adviceMessage) {
+					IntegrationMessageHeaderAccessor headerAccessor = new IntegrationMessageHeaderAccessor(message);
+					headerAccessor.setHeader(HEADER_NAME, messageHistory);
+					message = new AdviceMessage<>(message.getPayload(), headerAccessor.toMessageHeaders(),
+							adviceMessage.getInputMessage());
+				}
+				else {
+					if (!(message instanceof GenericMessage) &&
+							(messageBuilderFactory instanceof DefaultMessageBuilderFactory ||
+									messageBuilderFactory instanceof MutableMessageBuilderFactory)
+							&& LOGGER.isWarnEnabled()) {
+
+						LOGGER.warn("MessageHistory rebuilds the message and produces the result of the [" +
+								messageBuilderFactory + "], not an instance of the provided type [" +
+								message.getClass() + "]. Consider to supply a custom MessageBuilderFactory " +
+								"to retain custom messages during MessageHistory tracking.");
+					}
+					message = messageBuilderFactory.fromMessage(message)
+							.setHeader(HEADER_NAME, messageHistory)
+							.build();
 				}
-				message = messageBuilderFactory.fromMessage(message)
-						.setHeader(HEADER_NAME, history)
-						.build();
 			}
 		}
 		return message;
@@ -216,15 +219,19 @@ public int lastIndexOf(Object o) {
 		return this.components.lastIndexOf(o);
 	}
 
+	@Override
+	public Object clone() {
+		return new MessageHistory(new ArrayList<>(this.components));
+	}
+
 	@Override
 	public boolean equals(Object o) {
 		if (this == o) {
 			return true;
 		}
-		if (!(o instanceof MessageHistory)) {
+		if (!(o instanceof MessageHistory that)) {
 			return false;
 		}
-		MessageHistory that = (MessageHistory) o;
 		return this.components.equals(that.components);
 	}
 
@@ -318,6 +325,7 @@ private static Properties extractMetadata(NamedComponent component) {
 	 */
 	public static class Entry extends Properties {
 
+		@Serial
 		private static final long serialVersionUID = -8225834391885601079L;
 
 		public String getName() {

spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -27,6 +27,8 @@
 import org.springframework.integration.IntegrationPatternType;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.history.MessageHistory;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.management.IntegrationManagedResource;
 import org.springframework.jmx.export.annotation.ManagedResource;
 import org.springframework.messaging.Message;
@@ -193,19 +195,27 @@ protected void handleMessageInternal(Message<?> message) {
 		if (results != null) {
 			int sequenceSize = results.size();
 			int sequenceNumber = 1;
+			Message<?> messageToSend = message;
+			UUID correlationKey = message.getHeaders().getId();
+			boolean hasMessageHistory = message.getHeaders().containsKey(MessageHistory.HEADER_NAME) && sequenceSize > 1;
 			for (MessageChannel channel : results) {
-				final Message<?> messageToSend;
-				if (!this.applySequence) {
-					messageToSend = message;
-				}
-				else {
-					UUID id = message.getHeaders().getId();
-					messageToSend = getMessageBuilderFactory()
-							.fromMessage(message)
-							.pushSequenceDetails(id == null ? generateId() : id,
-									sequenceNumber++, sequenceSize)
-							.build();
+				if (this.applySequence || hasMessageHistory) {
+					AbstractIntegrationMessageBuilder<?> builder =
+							getMessageBuilderFactory()
+									.fromMessage(message);
+
+					if (this.applySequence) {
+						builder.pushSequenceDetails(correlationKey == null ? generateId() : correlationKey,
+								sequenceNumber++, sequenceSize);
+					}
+
+					if (hasMessageHistory) {
+						builder.cloneMessageHistoryIfAny();
+					}
+
+					messageToSend = builder.build();
 				}
+
 				if (channel != null) {
 					sent |= doSend(channel, messageToSend);
 				}

spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -35,6 +35,7 @@
 import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.DiscardingMessageHandler;
+import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.json.JacksonPresent;
 import org.springframework.integration.util.FunctionIterator;
@@ -276,10 +277,14 @@ private AbstractIntegrationMessageBuilder<?> createBuilder(Object item, Map<Stri
 			Object correlationId, int sequenceNumber, int sequenceSize) {
 
 		AbstractIntegrationMessageBuilder<?> builder = messageBuilderForReply(item);
-		builder.copyHeadersIfAbsent(headers);
+		builder.setHeader(MessageHistory.HEADER_NAME, headers.get(MessageHistory.HEADER_NAME))
+				.copyHeadersIfAbsent(headers)
+				.cloneMessageHistoryIfAny();
+
 		if (this.applySequence) {
 			builder.pushSequenceDetails(correlationId, sequenceNumber, sequenceSize);
 		}
+
 		return builder;
 	}