From 129ebdc625da516f142ceaec2bfe4eaad2bc8afa Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 12 Jan 2017 18:40:32 -0500 Subject: [PATCH] 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. --- .../AbstractMessageProducingHandler.java | 8 ++- .../dsl/flows/IntegrationFlowTests.java | 49 ++++++++++++++++++- src/reference/asciidoc/service-activator.adoc | 3 +- 3 files changed, 56 insertions(+), 4 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index d1e45a8308..505d6d0eb1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/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(); + } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java index 04b7561896..ba6504f3ac 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java +++ b/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()) + .transform(MessagingException::getFailedMessage) + .get(); + + } + + } + + @MessagingGateway + private interface ErrorRecovererFlowGateway { + + String testIt(String payload); + } @Service diff --git a/src/reference/asciidoc/service-activator.adoc b/src/reference/asciidoc/service-activator.adoc index 649473f44d..0eaf5da03a 100644 --- a/src/reference/asciidoc/service-activator.adoc +++ b/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`.