INT-4207: Fallback for replyChannel Resolution

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

Enhance `AbstractMessageProducingHandler` to fallback for `replyChannel` to the `reply` if it is `Message`.
That lets to avoid extra `bridge` configuration afterwards to make that `reply` as `request` for the same `replyChannel` resolution.

This situation happens in case of error handling when the request message is `ErrorMessage`, typically without original headers to properly consult.
But at the same time `failedMessage` in the `MessagingException` has all required headers.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -146,7 +146,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 		if (getOutputChannel() == null) {
 			Map<?, ?> routingSlipHeader = requestHeaders.get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class);
 			if (routingSlipHeader != null) {
-				Assert.isTrue(routingSlipHeader.size() == 1, "The RoutingSlip header value must be a SingletonMap");
+				Assert.isTrue(routingSlipHeader.size() == 1,
+						"The RoutingSlip header value must be a SingletonMap");
 				Object key = routingSlipHeader.keySet().iterator().next();
 				Object value = routingSlipHeader.values().iterator().next();
 				Assert.isInstanceOf(List.class, key, "The RoutingSlip key must be List");
@@ -174,6 +175,9 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 
 			if (replyChannel == null) {
 				replyChannel = requestHeaders.getReplyChannel();
+				if (replyChannel == null && reply instanceof Message) {
+					replyChannel = ((Message<?>) reply).getHeaders().getReplyChannel();
+				}
 			}
 		}
 

spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java

@@ -49,6 +49,7 @@ import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.annotation.MessageEndpoint;
 import org.springframework.integration.annotation.MessagingGateway;
 import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
@@ -58,7 +59,10 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.Pollers;
 import org.springframework.integration.dsl.channel.MessageChannels;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.GenericHandler;
+import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
 import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
+import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.SimpleMessageStore;
@@ -71,6 +75,7 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.ErrorMessage;
@@ -80,7 +85,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.stereotype.Component;
 import org.springframework.stereotype.Service;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringRunner;
 
 /**
@@ -415,6 +419,13 @@ public class IntegrationFlowTests {
 		assertEquals(6, receive3.getPayload());
 	}
 
+	@Autowired
+	private ErrorRecovererFlowGateway errorRecovererFlowGateway;
+
+	@Test
+	public void testReplyChannelFromReplyMessage() {
+		assertEquals("foo", this.errorRecovererFlowGateway.testIt("foo"));
+	}
 
 	@MessagingGateway
 	public interface ControlBusGateway {
@@ -681,6 +692,42 @@ public class IntegrationFlowTests {
 			return MessageChannels.queue().get();
 		}
 
+		@Bean
+		public IntegrationFlow errorRecovererFlow() {
+			return IntegrationFlows.from(ErrorRecovererFlowGateway.class)
+					.handle((GenericHandler<?>) (p, h) -> {
+						throw new RuntimeException("intentional");
+					}, e -> e.advice(retryAdvice()))
+					.get();
+		}
+
+		@Bean
+		public RequestHandlerRetryAdvice retryAdvice() {
+			RequestHandlerRetryAdvice requestHandlerRetryAdvice = new RequestHandlerRetryAdvice();
+			requestHandlerRetryAdvice.setRecoveryCallback(new ErrorMessageSendingRecoverer(recoveryChannel()));
+			return requestHandlerRetryAdvice;
+		}
+
+		@Bean
+		public MessageChannel recoveryChannel() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		public IntegrationFlow recoveryFlow() {
+			return IntegrationFlows.from(recoveryChannel())
+					.<MessagingException, Message>transform(MessagingException::getFailedMessage)
+					.get();
+
+		}
+
+	}
+
+	@MessagingGateway
+	private interface ErrorRecovererFlowGateway {
+
+		String testIt(String payload);
+
 	}
 
 	@Service

src/reference/asciidoc/service-activator.adoc

@@ -34,12 +34,13 @@ To determine the reply channel, it will first check if an "output-channel" was p
                        ref="somePojo" method="someMethod"/>
 ----
 
-If the method returns a result and no "output-channel" is defined, the framework will then check the Message's `replyChannel` header value.
+If the method returns a result and no "output-channel" is defined, the framework will then check the request Message's `replyChannel` header value.
 If that value is available, it will then check its type.
 If it is a`MessageChannel`, the reply message will be sent to that channel.
 If it is a `String`, then the endpoint will attempt to resolve the channel name to a channel instance.
 If the channel cannot be resolved, then a `DestinationResolutionException` will be thrown.
 It it can be resolved, the Message will be sent there.
+If the request Message doesn't have `replyChannel` header and and  the `reply` object is a `Message`, its `replyChannel` header is consulted for a target destination.
 This is the technique used for Request Reply messaging in Spring Integration, and it is also an example of the Return Address pattern.
 
 If your method returns a result, and you want to discard it and end the flow, you should configure the `output-channel` to send to a `NullChannel`.