INT-2480: Add aggregate headers strategy

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

* Introduce `headers-function` option into the `aggregator` for merging
and computing headers for the output message based on the completed
group
* Implement a `DefaultAggregateHeadersFunction` and use it in the
`AbstractAggregatingMessageGroupProcessor` for default behavior with
possible injection for any other implementation
* Add `DelegatingMessageGroupProcessor` to wrap any other
`MessageGroupProcessor` implementations with possible usage of the
`headersFunction` if result is not a `Message` or `MessageBuilder`
* Make `AbstractCorrelatingMessageHandler.getOutputProcessor()` as
`public` rto give access to this option from the `AggregatorSpec` to
be able to inject a `headersFunction` in Java DSL configuration
* Add `AbstractIntegrationMessageBuilder.getHeader()` to get access to
some underlying header avoiding extra `Map` in case of `getHeaders()`
* Change a logic in the `AbstractMessageProducingHandler.produceOutput()`
to consult a `reply` for the `replyChannel` as well `routingSlip` header
if the `reply` is a `Message` or `MessageBuilder`
* Introduce a `AbstractMessageProducingHandler.messageBuilderForReply()`
and use it in `AbstractMessageSplitter` to avoid duplication
* Validate a new functionality in tests
* Fix `FileOutboundGatewayParserTests` to rely on the `TemporaryFolder`
to clean up test files after using

* JavaDocs for `DefaultAggregateHeadersFunction`
* Some `router.adoc` polishing

* Fix link to Reactor in the `router.adoc`

* Add docs for new `Function<MessageGroup, Map<String, Object>>` strategy

* Doc polishing.

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java

@@ -16,12 +16,8 @@
 
 package org.springframework.integration.aggregator;
 
-import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Objects;
-import java.util.Set;
+import java.util.function.Function;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -29,14 +25,12 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
-import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.DefaultMessageBuilderFactory;
 import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
 
 /**
@@ -56,6 +50,8 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
 
 	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR - final
 
+	private Function<MessageGroup, Map<String, Object>> headersFunction = new DefaultAggregateHeadersFunction();
+
 	private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
 
 	private boolean messageBuilderFactorySet;
@@ -67,6 +63,20 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
 		this.beanFactory = beanFactory;
 	}
 
+	/**
+	 * Specify a {@link Function} to map {@link MessageGroup} into composed headers for output message.
+	 * @param headersFunction the {@link Function} to use.
+	 * @since 5.2
+	 */
+	public void setHeadersFunction(Function<MessageGroup, Map<String, Object>> headersFunction) {
+		Assert.notNull(headersFunction, "'headersFunction' must not be null");
+		this.headersFunction = headersFunction;
+	}
+
+	protected Function<MessageGroup, Map<String, Object>> getHeadersFunction() {
+		return this.headersFunction;
+	}
+
 	protected MessageBuilderFactory getMessageBuilderFactory() {
 		if (!this.messageBuilderFactorySet) {
 			if (this.beanFactory != null) {
@@ -81,7 +91,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
 	public final Object processMessageGroup(MessageGroup group) {
 		Assert.notNull(group, "MessageGroup must not be null");
 		Map<String, Object> headers = aggregateHeaders(group);
-		Object payload = this.aggregatePayloads(group, headers);
+		Object payload = aggregatePayloads(group, headers);
 		AbstractIntegrationMessageBuilder<?> builder;
 		if (payload instanceof Message<?>) {
 			builder = getMessageBuilderFactory().fromMessage((Message<?>) payload);
@@ -104,40 +114,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
 	 * @return The aggregated headers.
 	 */
 	protected Map<String, Object> aggregateHeaders(MessageGroup group) {
-		Map<String, Object> aggregatedHeaders = new HashMap<>();
-		Set<String> conflictKeys = doAggregateHeaders(group, aggregatedHeaders);
-		for (String keyToRemove : conflictKeys) {
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) "
-						+ "in MessageGroup with correlation key: " + group.getGroupId());
-			}
-			aggregatedHeaders.remove(keyToRemove);
-		}
-		return aggregatedHeaders;
-	}
-
-	private Set<String> doAggregateHeaders(MessageGroup group, Map<String, Object> aggregatedHeaders) {
-		Set<String> conflictKeys = new HashSet<>();
-		for (Message<?> message : group.getMessages()) {
-			for (Entry<String, Object> entry : message.getHeaders().entrySet()) {
-				String key = entry.getKey();
-				if (MessageHeaders.ID.equals(key) || MessageHeaders.TIMESTAMP.equals(key)
-						|| IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(key)
-						|| IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(key)) {
-					continue;
-				}
-				Object value = entry.getValue();
-				if (!aggregatedHeaders.containsKey(key)) {
-					aggregatedHeaders.put(key, value);
-				}
-				else {
-					if (!Objects.equals(value, aggregatedHeaders.get(key))) {
-						conflictKeys.add(key);
-					}
-				}
-			}
-		}
-		return conflictKeys;
+		return getHeadersFunction().apply(group);
 	}
 
 	protected abstract Object aggregatePayloads(MessageGroup group, Map<String, Object> defaultHeaders);

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -217,6 +217,15 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		this.outputProcessor = outputProcessor;
 	}
 
+	/**
+	 * Return a configured {@link MessageGroupProcessor}.
+	 * @return the configured {@link MessageGroupProcessor}
+	 * @since 5.2
+	 */
+	public MessageGroupProcessor getOutputProcessor() {
+		return this.outputProcessor;
+	}
+
 	public void setDiscardChannel(MessageChannel discardChannel) {
 		Assert.notNull(discardChannel, "'discardChannel' cannot be null");
 		this.discardChannel = discardChannel;
@@ -372,10 +381,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		return this.expireGroupScheduledFutures;
 	}
 
-	protected MessageGroupProcessor getOutputProcessor() {
-		return this.outputProcessor;
-	}
-
 	protected CorrelationStrategy getCorrelationStrategy() {
 		return this.correlationStrategy;
 	}

spring-integration-core/src/main/java/org/springframework/integration/aggregator/DefaultAggregateHeadersFunction.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.aggregator;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Function;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+
+/**
+ * The {@link Function} implementation for a default headers merging in the aggregator
+ * component. It takes all the unique headers from all the messages in group and removes
+ * those which are conflicted: have different values from different messages.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.2
+ *
+ * @see AbstractAggregatingMessageGroupProcessor
+ */
+public class DefaultAggregateHeadersFunction implements Function<MessageGroup, Map<String, Object>> {
+
+	private static final Log LOGGER = LogFactory.getLog(DefaultAggregateHeadersFunction.class);
+
+	@Override
+	public Map<String, Object> apply(MessageGroup messageGroup) {
+		Map<String, Object> aggregatedHeaders = new HashMap<>();
+		Set<String> conflictKeys = doAggregateHeaders(messageGroup, aggregatedHeaders);
+		for (String keyToRemove : conflictKeys) {
+			if (LOGGER.isDebugEnabled()) {
+				LOGGER.debug("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) "
+						+ "in MessageGroup with correlation key: " + messageGroup.getGroupId());
+			}
+			aggregatedHeaders.remove(keyToRemove);
+		}
+		return aggregatedHeaders;
+	}
+
+	private Set<String> doAggregateHeaders(MessageGroup group, Map<String, Object> aggregatedHeaders) {
+		Set<String> conflictKeys = new HashSet<>();
+		for (Message<?> message : group.getMessages()) {
+			for (Map.Entry<String, Object> entry : message.getHeaders().entrySet()) {
+				String key = entry.getKey();
+				if (MessageHeaders.ID.equals(key)
+						|| MessageHeaders.TIMESTAMP.equals(key)
+						|| IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(key)
+						|| IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(key)) {
+					continue;
+				}
+				Object value = entry.getValue();
+				if (!aggregatedHeaders.containsKey(key)) {
+					aggregatedHeaders.put(key, value);
+				}
+				else {
+					if (!Objects.equals(value, aggregatedHeaders.get(key))) {
+						conflictKeys.add(key);
+					}
+				}
+			}
+		}
+		return conflictKeys;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/aggregator/DelegatingMessageGroupProcessor.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.aggregator;
+
+import java.util.Map;
+import java.util.function.Function;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.integration.support.DefaultMessageBuilderFactory;
+import org.springframework.integration.support.MessageBuilderFactory;
+import org.springframework.integration.support.utils.IntegrationUtils;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ * The {@link MessageGroupProcessor} implementation with delegation to the provided {@code delegate}
+ * and optional aggregation for headers.
+ * <p>
+ * Unlike {@link AbstractAggregatingMessageGroupProcessor} this processor checks a result
+ * of the {@code delegate} call and aggregates headers into the output only
+ * if the result is not a {@link Message} or {@link AbstractIntegrationMessageBuilder}.
+ * <p>
+ * This processor is used internally for wrapping provided non-standard {@link MessageGroupProcessor}
+ * when a aggregate headers {@link Function} is provided.
+ * For POJO method invoking or SpEL expression evaluation it is recommended to use an
+ * {@link AbstractAggregatingMessageGroupProcessor} implementations.
+ *
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.2
+ */
+public class DelegatingMessageGroupProcessor implements MessageGroupProcessor, BeanFactoryAware, Lifecycle {
+
+	private final MessageGroupProcessor delegate;
+
+	private final Function<MessageGroup, Map<String, Object>> headersFunction;
+
+	private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
+
+	private volatile boolean messageBuilderFactorySet;
+
+	private BeanFactory beanFactory;
+
+	public DelegatingMessageGroupProcessor(MessageGroupProcessor delegate,
+			Function<MessageGroup, Map<String, Object>> headersFunction) {
+
+		Assert.notNull(delegate, "'delegate' must not be null");
+		Assert.notNull(headersFunction, "'headersFunction' must not be null");
+		this.delegate = delegate;
+		this.headersFunction = headersFunction;
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+		if (this.delegate instanceof BeanFactoryAware) {
+			((BeanFactoryAware) this.delegate).setBeanFactory(beanFactory);
+		}
+	}
+
+	@Override
+	public Object processMessageGroup(MessageGroup group) {
+		Object result = this.delegate.processMessageGroup(group);
+		if (!(result instanceof Message<?>) && !(result instanceof AbstractIntegrationMessageBuilder)) {
+			result = getMessageBuilderFactory()
+					.withPayload(result)
+					.copyHeadersIfAbsent(this.headersFunction.apply(group));
+		}
+		return result;
+	}
+
+	private MessageBuilderFactory getMessageBuilderFactory() {
+		if (!this.messageBuilderFactorySet) {
+			if (this.beanFactory != null) {
+				this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
+			}
+			this.messageBuilderFactorySet = true;
+		}
+		return this.messageBuilderFactory;
+	}
+
+	@Override
+	public void start() {
+		if (this.delegate instanceof Lifecycle) {
+			((Lifecycle) this.delegate).start();
+		}
+	}
+
+	@Override
+	public void stop() {
+		if (this.delegate instanceof Lifecycle) {
+			((Lifecycle) this.delegate).stop();
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.delegate instanceof Lifecycle && ((Lifecycle) this.delegate).isRunning();
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java

@@ -18,11 +18,10 @@ package org.springframework.integration.aggregator;
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 
-import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.messaging.Message;
 
@@ -32,6 +31,8 @@ import org.springframework.messaging.Message;
  * @author Iwein Fuld
  * @author Dave Syer
  * @author Oleg Zhurakousky
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class ResequencingMessageGroupProcessor implements MessageGroupProcessor {
@@ -42,9 +43,9 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
 		Collection<Message<?>> messages = group.getMessages();
 
 		if (messages.size() > 0) {
-			List<Message<?>> sorted = new ArrayList<Message<?>>(messages);
-			Collections.sort(sorted, this.comparator);
-			ArrayList<Message<?>> partialSequence = new ArrayList<Message<?>>();
+			List<Message<?>> sorted = new ArrayList<>(messages);
+			sorted.sort(this.comparator);
+			ArrayList<Message<?>> partialSequence = new ArrayList<>();
 			int previousSequence = extractSequenceNumber(sorted.get(0));
 			int currentSequence = previousSequence;
 			for (Message<?> message : sorted) {
@@ -63,6 +64,7 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
 	}
 
 	private Integer extractSequenceNumber(Message<?> message) {
-		return new IntegrationMessageHeaderAccessor(message).getSequenceNumber();
+		return StaticMessageHeaderAccessor.getSequenceNumber(message);
 	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java

@@ -17,15 +17,20 @@
 package org.springframework.integration.config;
 
 import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
 
 import org.aopalliance.aop.Advice;
 
 import org.springframework.expression.Expression;
+import org.springframework.integration.aggregator.AbstractAggregatingMessageGroupProcessor;
 import org.springframework.integration.aggregator.AggregatingMessageHandler;
 import org.springframework.integration.aggregator.CorrelationStrategy;
+import org.springframework.integration.aggregator.DelegatingMessageGroupProcessor;
 import org.springframework.integration.aggregator.MessageGroupProcessor;
 import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
 import org.springframework.integration.aggregator.ReleaseStrategy;
+import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.support.locks.LockRegistry;
 import org.springframework.integration.support.management.AbstractMessageHandlerMetrics;
@@ -92,6 +97,8 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 
 	private Boolean releaseLockBeforeSend;
 
+	private Function<MessageGroup, Map<String, Object>> headersFunction;
+
 	public void setProcessorBean(Object processorBean) {
 		this.processorBean = processorBean;
 	}
@@ -181,6 +188,10 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 		this.releaseLockBeforeSend = releaseLockBeforeSend;
 	}
 
+	public void setHeadersFunction(Function<MessageGroup, Map<String, Object>> headersFunction) {
+		this.headersFunction = headersFunction;
+	}
+
 	@Override
 	protected AggregatingMessageHandler createHandler() {
 		MessageGroupProcessor outputProcessor;
@@ -195,28 +206,38 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 				outputProcessor = new MethodInvokingMessageGroupProcessor(this.processorBean, this.methodName);
 			}
 		}
+
+		if (this.headersFunction != null) {
+			if (outputProcessor instanceof AbstractAggregatingMessageGroupProcessor) {
+				((AbstractAggregatingMessageGroupProcessor) outputProcessor).setHeadersFunction(this.headersFunction);
+			}
+			else {
+				outputProcessor = new DelegatingMessageGroupProcessor(outputProcessor, this.headersFunction);
+			}
+		}
+
 		AggregatingMessageHandler aggregator = new AggregatingMessageHandler(outputProcessor);
 		JavaUtils.INSTANCE
-			.acceptIfNotNull(this.expireGroupsUponCompletion, aggregator::setExpireGroupsUponCompletion)
-			.acceptIfNotNull(this.sendTimeout, aggregator::setSendTimeout)
-			.acceptIfNotNull(this.outputChannelName, aggregator::setOutputChannelName)
-			.acceptIfNotNull(this.metrics, aggregator::configureMetrics)
-			.acceptIfNotNull(this.statsEnabled, aggregator::setStatsEnabled)
-			.acceptIfNotNull(this.countsEnabled, aggregator::setCountsEnabled)
-			.acceptIfNotNull(this.lockRegistry, aggregator::setLockRegistry)
-			.acceptIfNotNull(this.messageStore, aggregator::setMessageStore)
-			.acceptIfNotNull(this.correlationStrategy, aggregator::setCorrelationStrategy)
-			.acceptIfNotNull(this.releaseStrategy, aggregator::setReleaseStrategy)
-			.acceptIfNotNull(this.groupTimeoutExpression, aggregator::setGroupTimeoutExpression)
-			.acceptIfNotNull(this.forceReleaseAdviceChain, aggregator::setForceReleaseAdviceChain)
-			.acceptIfNotNull(this.taskScheduler, aggregator::setTaskScheduler)
-			.acceptIfNotNull(this.discardChannel, aggregator::setDiscardChannel)
-			.acceptIfNotNull(this.discardChannelName, aggregator::setDiscardChannelName)
-			.acceptIfNotNull(this.sendPartialResultOnExpiry, aggregator::setSendPartialResultOnExpiry)
-			.acceptIfNotNull(this.minimumTimeoutForEmptyGroups, aggregator::setMinimumTimeoutForEmptyGroups)
-			.acceptIfNotNull(this.expireGroupsUponTimeout, aggregator::setExpireGroupsUponTimeout)
-			.acceptIfNotNull(this.popSequence, aggregator::setPopSequence)
-			.acceptIfNotNull(this.releaseLockBeforeSend, aggregator::setReleaseLockBeforeSend);
+				.acceptIfNotNull(this.expireGroupsUponCompletion, aggregator::setExpireGroupsUponCompletion)
+				.acceptIfNotNull(this.sendTimeout, aggregator::setSendTimeout)
+				.acceptIfNotNull(this.outputChannelName, aggregator::setOutputChannelName)
+				.acceptIfNotNull(this.metrics, aggregator::configureMetrics)
+				.acceptIfNotNull(this.statsEnabled, aggregator::setStatsEnabled)
+				.acceptIfNotNull(this.countsEnabled, aggregator::setCountsEnabled)
+				.acceptIfNotNull(this.lockRegistry, aggregator::setLockRegistry)
+				.acceptIfNotNull(this.messageStore, aggregator::setMessageStore)
+				.acceptIfNotNull(this.correlationStrategy, aggregator::setCorrelationStrategy)
+				.acceptIfNotNull(this.releaseStrategy, aggregator::setReleaseStrategy)
+				.acceptIfNotNull(this.groupTimeoutExpression, aggregator::setGroupTimeoutExpression)
+				.acceptIfNotNull(this.forceReleaseAdviceChain, aggregator::setForceReleaseAdviceChain)
+				.acceptIfNotNull(this.taskScheduler, aggregator::setTaskScheduler)
+				.acceptIfNotNull(this.discardChannel, aggregator::setDiscardChannel)
+				.acceptIfNotNull(this.discardChannelName, aggregator::setDiscardChannelName)
+				.acceptIfNotNull(this.sendPartialResultOnExpiry, aggregator::setSendPartialResultOnExpiry)
+				.acceptIfNotNull(this.minimumTimeoutForEmptyGroups, aggregator::setMinimumTimeoutForEmptyGroups)
+				.acceptIfNotNull(this.expireGroupsUponTimeout, aggregator::setExpireGroupsUponTimeout)
+				.acceptIfNotNull(this.popSequence, aggregator::setPopSequence)
+				.acceptIfNotNull(this.releaseLockBeforeSend, aggregator::setReleaseLockBeforeSend);
 
 		return aggregator;
 	}

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java

@@ -38,6 +38,7 @@ import org.springframework.util.StringUtils;
  * @author Dave Syer
  * @author Stefan Ferstl
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser {
 
@@ -49,6 +50,7 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser {
 				parserContext);
 		String ref = element.getAttribute(REF_ATTRIBUTE);
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(AggregatorFactoryBean.class);
+		String headersFunction = element.getAttribute("headers-function");
 		BeanMetadataElement processor = null;
 
 		if (innerHandlerDefinition != null || StringUtils.hasText(ref)) {
@@ -59,18 +61,25 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser {
 				processor = new RuntimeBeanReference(ref);
 			}
 			builder.addPropertyValue("processorBean", processor);
+			if (StringUtils.hasText(headersFunction)) {
+				builder.addPropertyReference("headersFunction", headersFunction);
+			}
 		}
 		else {
+			BeanDefinitionBuilder groupProcessorBuilder;
 			if (StringUtils.hasText(element.getAttribute(EXPRESSION_ATTRIBUTE))) {
 				String expression = element.getAttribute(EXPRESSION_ATTRIBUTE);
-				BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
-						.genericBeanDefinition(ExpressionEvaluatingMessageGroupProcessor.class);
-				adapterBuilder.addConstructorArgValue(expression);
-				builder.addPropertyValue("processorBean", adapterBuilder.getBeanDefinition());
+				groupProcessorBuilder =
+						BeanDefinitionBuilder.genericBeanDefinition(ExpressionEvaluatingMessageGroupProcessor.class);
+				groupProcessorBuilder.addConstructorArgValue(expression);
 			}
 			else {
-				builder.addPropertyValue("processorBean", BeanDefinitionBuilder
-						.genericBeanDefinition(DefaultAggregatingMessageGroupProcessor.class).getBeanDefinition());
+				groupProcessorBuilder =
+						BeanDefinitionBuilder.genericBeanDefinition(DefaultAggregatingMessageGroupProcessor.class);
+			}
+			builder.addPropertyValue("processorBean", groupProcessorBuilder.getBeanDefinition());
+			if (StringUtils.hasText(headersFunction)) {
+				groupProcessorBuilder.addPropertyReference("headersFunction", headersFunction);
 			}
 		}
 
@@ -79,7 +88,7 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser {
 			builder.addPropertyValue("methodName", method);
 		}
 
-		this.doParse(builder, element, processor, parserContext);
+		doParse(builder, element, processor, parserContext);
 
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_COMPLETION);
 

spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java

@@ -16,11 +16,17 @@
 
 package org.springframework.integration.dsl;
 
+import java.util.Map;
+import java.util.function.Function;
+
+import org.springframework.integration.aggregator.AbstractAggregatingMessageGroupProcessor;
 import org.springframework.integration.aggregator.AggregatingMessageHandler;
 import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
+import org.springframework.integration.aggregator.DelegatingMessageGroupProcessor;
 import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor;
 import org.springframework.integration.aggregator.MessageGroupProcessor;
 import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
+import org.springframework.integration.store.MessageGroup;
 
 /**
  * A {@link CorrelationHandlerSpec} for an {@link AggregatingMessageHandler}.
@@ -31,6 +37,8 @@ import org.springframework.integration.aggregator.MethodInvokingMessageGroupProc
  */
 public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, AggregatingMessageHandler> {
 
+	private Function<MessageGroup, Map<String, Object>> headersFunction;
+
 	AggregatorSpec() {
 		super(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()));
 	}
@@ -59,9 +67,9 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
 	 */
 	public AggregatorSpec processor(Object target, String methodName) {
 		super.processor(target);
-		return this.outputProcessor(methodName != null
-									? new MethodInvokingMessageGroupProcessor(target, methodName)
-									: new MethodInvokingMessageGroupProcessor(target));
+		return outputProcessor(methodName != null
+				? new MethodInvokingMessageGroupProcessor(target, methodName)
+				: new MethodInvokingMessageGroupProcessor(target));
 	}
 
 	/**
@@ -71,7 +79,7 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
 	 * @return the aggregator spec.
 	 */
 	public AggregatorSpec outputExpression(String expression) {
-		return this.outputProcessor(new ExpressionEvaluatingMessageGroupProcessor(expression));
+		return outputProcessor(new ExpressionEvaluatingMessageGroupProcessor(expression));
 	}
 
 	/**
@@ -95,4 +103,33 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
 		return _this();
 	}
 
+	/**
+	 * Configure a {@link Function} to merge and compute headers for reply
+	 * based on the completed {@link MessageGroup}.
+	 * @param headersFunction the {@link Function} to merge and compute headers for reply
+	 * based on the completed {@link MessageGroup}.
+	 * @return the aggregator spec.
+	 * @since 5.2
+	 */
+	public AggregatorSpec headersFunction(Function<MessageGroup, Map<String, Object>> headersFunction) {
+		this.headersFunction = headersFunction;
+		return _this();
+	}
+
+
+	@Override
+	public Map<Object, String> getComponentsToRegister() {
+		if (this.headersFunction != null) {
+			MessageGroupProcessor outputProcessor = this.handler.getOutputProcessor();
+			if (outputProcessor instanceof AbstractAggregatingMessageGroupProcessor) {
+				((AbstractAggregatingMessageGroupProcessor) outputProcessor).setHeadersFunction(this.headersFunction);
+			}
+			else {
+				this.handler.setOutputProcessor(
+						new DelegatingMessageGroupProcessor(outputProcessor, this.headersFunction));
+			}
+		}
+		return super.getComponentsToRegister();
+	}
+
 }

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

@@ -158,9 +158,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 			this.notPropagatedHeaders = headerPatterns.toArray(new String[0]);
 		}
 
-		boolean hasAsterisk = headerPatterns.contains("*");
-
-		if (hasAsterisk) {
+		if (headerPatterns.contains("*")) {
 			this.notPropagatedHeaders = new String[] { "*" };
 			this.noHeadersPropagation = true;
 		}
@@ -240,12 +238,11 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 	}
 
 	protected void produceOutput(Object replyArg, final Message<?> requestMessage) {
-		final MessageHeaders requestHeaders = requestMessage.getHeaders();
-
+		MessageHeaders requestHeaders = requestMessage.getHeaders();
 		Object reply = replyArg;
 		Object replyChannel = null;
 		if (getOutputChannel() == null) {
-			Map<?, ?> routingSlipHeader = requestHeaders.get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class);
+			Map<?, ?> routingSlipHeader = obtainRoutingSlipHeader(requestHeaders, reply);
 			if (routingSlipHeader != null) {
 				Assert.isTrue(routingSlipHeader.size() == 1,
 						"The RoutingSlip header value must be a SingletonMap");
@@ -260,18 +257,45 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 					reply = addRoutingSlipHeader(reply, routingSlip, routingSlipIndex);
 				}
 			}
-
 			if (replyChannel == null) {
-				replyChannel = requestHeaders.getReplyChannel();
-				if (replyChannel == null && reply instanceof Message) {
-					replyChannel = ((Message<?>) reply).getHeaders().getReplyChannel();
-				}
+				replyChannel = obtainReplyChannel(requestHeaders, reply);
 			}
 		}
 		doProduceOutput(requestMessage, requestHeaders, reply, replyChannel);
 	}
 
-	private void doProduceOutput(final Message<?> requestMessage, final MessageHeaders requestHeaders, Object reply,
+	@Nullable
+	private Map<?, ?> obtainRoutingSlipHeader(MessageHeaders requestHeaders, Object reply) {
+		Map<?, ?> routingSlipHeader = requestHeaders.get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class);
+		if (routingSlipHeader == null) {
+			if (reply instanceof Message) {
+				routingSlipHeader = ((Message<?>) reply).getHeaders()
+						.get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class);
+			}
+			else if (reply instanceof AbstractIntegrationMessageBuilder<?>) {
+				routingSlipHeader = ((AbstractIntegrationMessageBuilder<?>) reply)
+						.getHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class);
+			}
+		}
+		return routingSlipHeader;
+	}
+
+	@Nullable
+	private Object obtainReplyChannel(MessageHeaders requestHeaders, Object reply) {
+		Object replyChannel = requestHeaders.getReplyChannel();
+		if (replyChannel == null) {
+			if (reply instanceof Message) {
+				replyChannel = ((Message<?>) reply).getHeaders().getReplyChannel();
+			}
+			else if (reply instanceof AbstractIntegrationMessageBuilder<?>) {
+				replyChannel = ((AbstractIntegrationMessageBuilder<?>) reply)
+						.getHeader(MessageHeaders.REPLY_CHANNEL, Object.class);
+			}
+		}
+		return replyChannel;
+	}
+
+	private void doProduceOutput(Message<?> requestMessage, MessageHeaders requestHeaders, Object reply,
 			Object replyChannel) {
 
 		if (this.async && (reply instanceof ListenableFuture<?> || reply instanceof Publisher<?>)) {
@@ -296,19 +320,22 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 	private AbstractIntegrationMessageBuilder<?> addRoutingSlipHeader(Object reply, List<?> routingSlip,
 			AtomicInteger routingSlipIndex) {
 
-		//TODO Migrate to the SF MessageBuilder
-		AbstractIntegrationMessageBuilder<?> builder = null;
+		return messageBuilderForReply(reply)
+				.setHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP,
+						Collections.singletonMap(routingSlip, routingSlipIndex.get()));
+	}
+
+	protected AbstractIntegrationMessageBuilder<?> messageBuilderForReply(Object reply) {
+		AbstractIntegrationMessageBuilder<?> builder;
 		if (reply instanceof Message) {
-			builder = this.getMessageBuilderFactory().fromMessage((Message<?>) reply);
+			builder = getMessageBuilderFactory().fromMessage((Message<?>) reply);
 		}
 		else if (reply instanceof AbstractIntegrationMessageBuilder) {
 			builder = (AbstractIntegrationMessageBuilder<?>) reply;
 		}
 		else {
-			builder = this.getMessageBuilderFactory().withPayload(reply);
+			builder = getMessageBuilderFactory().withPayload(reply);
 		}
-		builder.setHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP,
-				Collections.singletonMap(routingSlip, routingSlipIndex.get()));
 		return builder;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/routingslip/ExpressionEvaluatingRoutingSlipRouteStrategy.java

@@ -59,6 +59,7 @@ import org.springframework.messaging.Message;
  *
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 4.1
  */
 public class ExpressionEvaluatingRoutingSlipRouteStrategy
@@ -98,8 +99,7 @@ public class ExpressionEvaluatingRoutingSlipRouteStrategy
 
 	@Override
 	public Object getNextPath(Message<?> requestMessage, Object reply) {
-		return this.expression.getValue(this.evaluationContext, new RequestAndReply(requestMessage, reply),
-				String.class);
+		return this.expression.getValue(this.evaluationContext, new RequestAndReply(requestMessage, reply));
 	}
 
 	@Override

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

@@ -263,16 +263,7 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 	private AbstractIntegrationMessageBuilder<?> createBuilder(Object item, Map<String, Object> headers,
 			Object correlationId, int sequenceNumber, int sequenceSize) {
 
-		AbstractIntegrationMessageBuilder<?> builder;
-		if (item instanceof Message) {
-			builder = getMessageBuilderFactory().fromMessage((Message<?>) item);
-		}
-		else if (item instanceof AbstractIntegrationMessageBuilder) {
-			builder = (AbstractIntegrationMessageBuilder<?>) item;
-		}
-		else {
-			builder = getMessageBuilderFactory().withPayload(item);
-		}
+		AbstractIntegrationMessageBuilder<?> builder = messageBuilderForReply(item);
 		builder.copyHeadersIfAbsent(headers);
 		if (this.applySequence) {
 			builder.pushSequenceDetails(correlationId, sequenceNumber, sequenceSize);

spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java

@@ -176,6 +176,9 @@ public abstract class AbstractIntegrationMessageBuilder<T> {
 
 	public abstract Map<String, Object> getHeaders();
 
+	@Nullable
+	public abstract <V> V getHeader(String key, Class<V> type);
+
 	/**
 	 * Set the value for the given header name. If the provided value is <code>null</code>, the header will be removed.
 	 * @param headerName The header name.

spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java

@@ -64,7 +64,7 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 	/**
 	 * Private constructor to be invoked from the static factory methods only.
 	 */
-	private MessageBuilder(T payload, Message<T> originalMessage) {
+	private MessageBuilder(T payload, @Nullable Message<T> originalMessage) {
 		Assert.notNull(payload, "payload must not be null");
 		this.payload = payload;
 		this.originalMessage = originalMessage;
@@ -84,6 +84,12 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 		return this.headerAccessor.toMap();
 	}
 
+	@Nullable
+	@Override
+	public <V> V getHeader(String key, Class<V> type) {
+		return this.headerAccessor.getHeader(key, type);
+	}
+
 	/**
 	 * Create a builder for a new {@link Message} instance pre-populated with all of the headers copied from the
 	 * provided message. The payload of the provided Message will also be used as the payload for the new message.
@@ -94,23 +100,21 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 	 */
 	public static <T> MessageBuilder<T> fromMessage(Message<T> message) {
 		Assert.notNull(message, "message must not be null");
-		return new MessageBuilder<T>(message.getPayload(), message);
+		return new MessageBuilder<>(message.getPayload(), message);
 	}
 
 	/**
 	 * Create a builder for a new {@link Message} instance with the provided payload.
-	 *
 	 * @param payload the payload for the new message
 	 * @param <T> The type of the payload.
 	 * @return A MessageBuilder.
 	 */
 	public static <T> MessageBuilder<T> withPayload(T payload) {
-		return new MessageBuilder<T>(payload, null);
+		return new MessageBuilder<>(payload, null);
 	}
 
 	/**
 	 * Set the value for the given header name. If the provided value is <code>null</code>, the header will be removed.
-	 *
 	 * @param headerName The header name.
 	 * @param headerValue The header value.
 	 * @return this MessageBuilder.
@@ -123,7 +127,6 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 
 	/**
 	 * Set the value for the given header name only if the header name is not already associated with a value.
-	 *
 	 * @param headerName The header name.
 	 * @param headerValue The header value.
 	 * @return this MessageBuilder.
@@ -138,7 +141,6 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 	 * Removes all headers provided via array of 'headerPatterns'. As the name suggests the array
 	 * may contain simple matching patterns for header names. Supported pattern styles are:
 	 * "xxx*", "*xxx", "*xxx*" and "xxx*yyy".
-	 *
 	 * @param headerPatterns The header patterns.
 	 * @return this MessageBuilder.
 	 */
@@ -168,10 +170,8 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 	 * Copy the name-value pairs from the provided Map. This operation will overwrite any existing values. Use {
 	 * {@link #copyHeadersIfAbsent(Map)} to avoid overwriting values. Note that the 'id' and 'timestamp' header values
 	 * will never be overwritten.
-	 *
 	 * @param headersToCopy The headers to copy.
 	 * @return this MessageBuilder.
-	 *
 	 * @see MessageHeaders#ID
 	 * @see MessageHeaders#TIMESTAMP
 	 */
@@ -183,7 +183,6 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 
 	/**
 	 * Copy the name-value pairs from the provided Map. This operation will <em>not</em> overwrite any existing values.
-	 *
 	 * @param headersToCopy The headers to copy.
 	 * @return this MessageBuilder.
 	 */
@@ -225,8 +224,8 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 
 	/*
 	 * The following overrides (delegating to super) are provided to ease the
-	 *  pain for existing applications that use the builder API and expect
-	 *  a MessageBuilder to be returned.
+	 * pain for existing applications that use the builder API and expect
+	 * a MessageBuilder to be returned.
 	 */
 	@Override
 	public MessageBuilder<T> pushSequenceDetails(Object correlationId, int sequenceNumber, int sequenceSize) {
@@ -320,12 +319,13 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 	public Message<T> build() {
 		if (!this.modified && !this.headerAccessor.isModified() && this.originalMessage != null
 				&& !containsReadOnly(this.originalMessage.getHeaders())) {
+
 			return this.originalMessage;
 		}
 		if (this.payload instanceof Throwable) {
 			return (Message<T>) new ErrorMessage((Throwable) this.payload, this.headerAccessor.toMap());
 		}
-		return new GenericMessage<T>(this.payload, this.headerAccessor.toMap());
+		return new GenericMessage<>(this.payload, this.headerAccessor.toMap());
 	}
 
 	private boolean containsReadOnly(MessageHeaders headers) {
@@ -339,5 +339,4 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
 		return false;
 	}
 
-
 }

spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java

@@ -43,6 +43,7 @@ import org.springframework.util.StringUtils;
  *
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 4.0
  *
  */
@@ -76,6 +77,21 @@ public final class MutableMessageBuilder<T> extends AbstractIntegrationMessageBu
 		return this.headers;
 	}
 
+	@SuppressWarnings("unchecked")
+	@Nullable
+	@Override
+	public <V> V getHeader(String key, Class<V> type) {
+		Object value = this.headers.get(key);
+		if (value == null) {
+			return null;
+		}
+		if (!type.isAssignableFrom(value.getClass())) {
+			throw new IllegalArgumentException("Incorrect type specified for header '" + key + "'. Expected [" + type
+					+ "] but actual type is [" + value.getClass() + "]");
+		}
+		return (V) value;
+	}
+
 	/**
 	 * Create a builder for a new {@link Message} instance with the provided payload.
 	 * @param payload the payload for the new message
@@ -143,7 +159,7 @@ public final class MutableMessageBuilder<T> extends AbstractIntegrationMessageBu
 
 	@Override
 	public AbstractIntegrationMessageBuilder<T> removeHeaders(String... headerPatterns) {
-		List<String> headersToRemove = new ArrayList<String>();
+		List<String> headersToRemove = new ArrayList<>();
 		for (String pattern : headerPatterns) {
 			if (StringUtils.hasLength(pattern)) {
 				if (pattern.contains("*")) {
@@ -161,7 +177,7 @@ public final class MutableMessageBuilder<T> extends AbstractIntegrationMessageBu
 	}
 
 	private List<String> getMatchingHeaderNames(String pattern, Map<String, Object> headers) {
-		List<String> matchingHeaderNames = new ArrayList<String>();
+		List<String> matchingHeaderNames = new ArrayList<>();
 		if (headers != null) {
 			for (Map.Entry<String, Object> header : headers.entrySet()) {
 				if (PatternMatchUtils.simpleMatch(pattern, header.getKey())) {

spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd

@@ -3724,6 +3724,19 @@
 						<xsd:union memberTypes="xsd:boolean xsd:string" />
 					</xsd:simpleType>
 				</xsd:attribute>
+				<xsd:attribute name="headers-function" type="xsd:string">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type type="java.util.function.Function" />
+							</tool:annotation>
+						</xsd:appinfo>
+						<xsd:documentation>
+							A reference to the 'Function' for merging and computing message headers for reply
+							based on the 'MessageGroup' to release.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml

@@ -11,7 +11,12 @@
 		<queue capacity="5" />
 	</channel>
 
-	<aggregator ref="summer" method="sum" input-channel="input" output-channel="output" expression="">
+	<beans:bean id="headersFunction"
+				class="org.springframework.integration.aggregator.integration.AggregatorIntegrationTests"
+				factory-method="firstMessageHeaders"/>
+
+	<aggregator ref="summer" method="sum" input-channel="input" output-channel="output" expression=""
+				headers-function="headersFunction">
 		<poller task-executor="executor" max-messages-per-poll="5" fixed-delay="20" />
 	</aggregator>
 

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java

@@ -24,6 +24,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
+import java.util.function.Function;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -32,6 +33,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
@@ -92,6 +94,7 @@ public class AggregatorIntegrationTests {
 		Message<?> receive = output.receive(10000);
 		assertThat(receive).isNotNull();
 		assertThat(receive.getPayload()).isEqualTo(1 + 2 + 3 + 4);
+		assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)).isEqualTo(0);
 	}
 
 	@Test
@@ -245,6 +248,18 @@ public class AggregatorIntegrationTests {
 	}
 
 	// configured in context associated with this test
+	private Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) {
+		Map<String, Object> headers = new HashMap<>();
+		headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber);
+		headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize);
+		headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId);
+		return headers;
+	}
+
+	public static Function<MessageGroup, Map<String, Object>> firstMessageHeaders() {
+		return (messageGroup) -> messageGroup.getOne().getHeaders();
+	}
+
 	public static class SummingAggregator {
 		public Integer sum(List<Integer> numbers) {
 			int result = 0;
@@ -255,12 +270,6 @@ public class AggregatorIntegrationTests {
 		}
 	}
 
-	private Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) {
-		Map<String, Object> headers = new HashMap<>();
-		headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber);
-		headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize);
-		headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId);
-		return headers;
-	}
+
 
 }

spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml

@@ -83,7 +83,7 @@
 	<channel id="fooChannel"/>
 
 	<header-enricher input-channel="routingSlipInput">
-		<routing-slip value="fooChannel; barExpression; bazRoutingSlip"/>
+		<routing-slip value="request.headers.replyChannel; fooChannel; barExpression; bazRoutingSlip"/>
 	</header-enricher>
 
 	<header-enricher input-channel="payloadExpressionInput">

spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java

@@ -270,9 +270,10 @@ public class HeaderEnricherTests {
 		@SuppressWarnings("unchecked")
 		List<Object> routingSlipPath = (List<Object>) ((Map<?, ?>) routingSlip).keySet().iterator().next();
 
-		assertThat(routingSlipPath.get(0)).isEqualTo("fooChannel");
-		assertThat(routingSlipPath.get(1)).isInstanceOf(ExpressionEvaluatingRoutingSlipRouteStrategy.class);
-		assertThat(routingSlipPath.get(2)).isEqualTo("bazRoutingSlip");
+		assertThat(routingSlipPath.get(0)).isInstanceOf(ExpressionEvaluatingRoutingSlipRouteStrategy.class);
+		assertThat(routingSlipPath.get(1)).isEqualTo("fooChannel");
+		assertThat(routingSlipPath.get(2)).isInstanceOf(ExpressionEvaluatingRoutingSlipRouteStrategy.class);
+		assertThat(routingSlipPath.get(3)).isEqualTo("bazRoutingSlip");
 	}
 
 

spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -141,9 +142,10 @@ public class CorrelationHandlerTests {
 	public void testSubscriberAggregateFlow() {
 		this.subscriberAggregateFlowInput.send(new GenericMessage<>("test"));
 
-		Message<?> receive1 = this.subscriberAggregateResult.receive(10000);
-		assertThat(receive1).isNotNull();
-		assertThat(receive1.getPayload()).isEqualTo("Hello World!");
+		Message<?> receive = this.subscriberAggregateResult.receive(10000);
+		assertThat(receive).isNotNull();
+		assertThat(receive.getPayload()).isEqualTo("Hello World!");
+		assertThat(receive.getHeaders().get("foo")).isEqualTo("bar");
 	}
 
 
@@ -274,10 +276,13 @@ public class CorrelationHandlerTests {
 		@Bean
 		public IntegrationFlow publishSubscribeAggregateFlow() {
 			return flow -> flow
-					.aggregate(a -> a.outputProcessor(g -> g.getMessages()
-							.stream()
-							.map(m -> (String) m.getPayload())
-							.collect(Collectors.joining(" "))))
+					.aggregate(a -> a
+							.outputProcessor((group) -> group
+									.getMessages()
+									.stream()
+									.map(m -> (String) m.getPayload())
+									.collect(Collectors.joining(" ")))
+							.headersFunction((group) -> Collections.singletonMap("foo", "bar")))
 					.channel(MessageChannels.queue("subscriberAggregateResult"));
 		}