INT-4457: Make log() in the end as reply-aware

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

When an `IntegrationFlow` is reply-based (we expect a reply in the
beginning) and `log()` (or `wireTap()`) is used in the end, we are
forced to add an empty `bridge()` in the end to ensure a `replyChannel`
header routing

* Fix `IntegrationFlowDefinition` to add `enrichHeaders()` in the end
to populate a `nullChannel` as a `replyChannel` header if that is missed
in the request message headers.
This way we cover both use-cases when we expect reply from the flow and
when it is used as a one-way scenario
* Improve a `HeaderEnricher` do not create a new `Message` if there are
no new headers to add/remove
* Remove a note from the `dsl.adoc` about now redundant `bridge()` after
`log()`
* Resolve TODO in the `.handle()` paragraph

Author: artembilan

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.dsl;
 
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -92,6 +93,7 @@
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
@@ -2740,7 +2742,9 @@ else if (this.currentMessageChannel != null) {
 								.stream()
 								.reduce((first, second) -> second);
 				if (lastComponent.get() instanceof WireTapSpec) {
-					channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME);
+					enrichHeaders(
+							Collections.singletonMap(
+									MessageHeaders.REPLY_CHANNEL, IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME));
 				}
 			}
 

spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderEnricher.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.
@@ -29,11 +29,11 @@
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.handler.MessageProcessor;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.DefaultMessageBuilderFactory;
 import org.springframework.integration.transformer.support.HeaderValueMessageProcessor;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
-import org.springframework.messaging.MessagingException;
 
 /**
  * A Transformer that adds statically configured header values to a Message.
@@ -64,12 +64,13 @@ public HeaderEnricher() {
 
 	/**
 	 * Create a HeaderEnricher with the given map of headers.
-	 *
 	 * @param headersToAdd The headers to add.
 	 */
 	public HeaderEnricher(Map<String, ? extends HeaderValueMessageProcessor<?>> headersToAdd) {
-		this.headersToAdd = (headersToAdd != null) ? headersToAdd
-				: new HashMap<String, HeaderValueMessageProcessor<Object>>();
+		this.headersToAdd =
+				headersToAdd != null
+						? headersToAdd
+						: new HashMap<String, HeaderValueMessageProcessor<Object>>();
 	}
 
 	public <T> void setMessageProcessor(MessageProcessor<T> messageProcessor) {
@@ -86,7 +87,6 @@ public void setDefaultOverwrite(boolean defaultOverwrite) {
 	 * <code>true</code>. Set this to <code>false</code> if a
 	 * <code>null</code> value should trigger <i>removal</i> of the
 	 * corresponding header instead.
-	 *
 	 * @param shouldSkipNulls true when null values should be skipped.
 	 */
 	public void setShouldSkipNulls(boolean shouldSkipNulls) {
@@ -96,52 +96,56 @@ public void setShouldSkipNulls(boolean shouldSkipNulls) {
 
 	@Override
 	public String getComponentType() {
-		return "transformer"; // backwards compatibility
+		return "header-enricher";
 	}
 
 	@Override
 	public Message<?> transform(Message<?> message) {
-		try {
-			Map<String, Object> headerMap = new HashMap<String, Object>(message.getHeaders());
-			this.addHeadersFromMessageProcessor(message, headerMap);
-			for (Map.Entry<String, ? extends HeaderValueMessageProcessor<?>> entry : this.headersToAdd.entrySet()) {
-				String key = entry.getKey();
-				HeaderValueMessageProcessor<?> valueProcessor = entry.getValue();
-
-				Boolean shouldOverwrite = valueProcessor.isOverwrite();
-				if (shouldOverwrite == null) {
-					shouldOverwrite = this.defaultOverwrite;
-				}
+		MessageHeaders messageHeaders = message.getHeaders();
 
-				boolean headerDoesNotExist = headerMap.get(key) == null;
+		AbstractIntegrationMessageBuilder<?> messageBuilder =
+				getMessageBuilderFactory()
+						.fromMessage(message);
 
-				/*
-				 * Only evaluate value expression if necessary
-				 */
-				if (headerDoesNotExist || shouldOverwrite) {
-					Object value = valueProcessor.processMessage(message);
-					if (value != null || !this.shouldSkipNulls) {
-						headerMap.put(key, value);
-					}
+		addHeadersFromMessageProcessor(message, messageBuilder);
+		for (Map.Entry<String, ? extends HeaderValueMessageProcessor<?>> entry : this.headersToAdd.entrySet()) {
+			String key = entry.getKey();
+			HeaderValueMessageProcessor<?> valueProcessor = entry.getValue();
+
+			Boolean shouldOverwrite = valueProcessor.isOverwrite();
+			if (shouldOverwrite == null) {
+				shouldOverwrite = this.defaultOverwrite;
+			}
+
+			boolean headerDoesNotExist = messageHeaders.get(key) == null;
+
+			/*
+			 * Only evaluate value expression if necessary
+			 */
+			if (headerDoesNotExist || shouldOverwrite) {
+				Object value = valueProcessor.processMessage(message);
+				if (value != null || !this.shouldSkipNulls) {
+					messageBuilder.setHeader(key, value);
 				}
 			}
-			return this.getMessageBuilderFactory().withPayload(message.getPayload()).copyHeaders(headerMap).build();
-		}
-		catch (Exception e) {
-			throw new MessagingException(message, "failed to transform message headers", e);
 		}
+
+		return messageBuilder.build();
 	}
 
-	private void addHeadersFromMessageProcessor(Message<?> message, Map<String, Object> headerMap) {
+	private void addHeadersFromMessageProcessor(Message<?> message,
+			AbstractIntegrationMessageBuilder<?> messageBuilder) {
+
 		if (this.messageProcessor != null) {
 			Object result = this.messageProcessor.processMessage(message);
 			if (result instanceof Map) {
+				MessageHeaders messageHeaders = message.getHeaders();
 				Map<?, ?> resultMap = (Map<?, ?>) result;
 				for (Entry<?, ?> entry : resultMap.entrySet()) {
 					Object key = entry.getKey();
 					if (key instanceof String) {
-						if (this.defaultOverwrite || headerMap.get(key) == null) {
-							headerMap.put((String) key, entry.getValue());
+						if (this.defaultOverwrite || messageHeaders.get(key) == null) {
+							messageBuilder.setHeader((String) key, entry.getValue());
 						}
 					}
 					else if (logger.isDebugEnabled()) {

spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java

@@ -133,8 +133,7 @@ public static class ContextConfiguration {
 		@Bean
 		public IntegrationFlow testGateway() {
 			return f -> f.gateway("processChannel", g -> g.replyChannel("replyChannel"))
-					.log()
-					.bridge();
+					.log();
 		}
 
 		@Bean