Fix MessagingGatewaySupport for reactive error (#3319)

artembilan · garyrussell · commit 9082c12ba18b · 2020-07-01T15:54:24.000-04:00
* Fix MessagingGatewaySupport for reactive error

The `onErrorResume()` was in a wrong place for the
`doSendAndReceiveMessageReactive()`: we have to catch all the exceptions
from the top level `Mono`, not only a reply one as it was before.

Ensure in HTTP and WebFlux test that behavior is fixed

**Cherry-pick to `5.3.x` &amp; `5.2.x`**

* * Remove unused imports

Co-authored-by: Artem Bilan &lt;abilan@vmware.com&gt;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
@@ -613,39 +613,40 @@ protected Mono<Message<?>> sendAndReceiveMessageReactive(Object object) {
 
 	private Mono<Message<?>> doSendAndReceiveMessageReactive(MessageChannel requestChannel, Object object,
 			boolean error) {
+		final Message<?> requestMessage;
+		try {
+			Message<?> message =
+					object instanceof Message<?>
+							? (Message<?>) object
+							: this.requestMapper.toMessage(object);
 
-		return Mono.defer(() -> {
-			Message<?> message;
-			try {
-				message = object instanceof Message<?>
-						? (Message<?>) object
-						: this.requestMapper.toMessage(object);
-
-				message = this.historyWritingPostProcessor.postProcessMessage(message);
+			message = this.historyWritingPostProcessor.postProcessMessage(message);
+			requestMessage = message;
+		}
+		catch (Exception e) {
+			throw new MessageMappingException("Cannot map to message: " + object, e);
+		}
 
-			}
-			catch (Exception e) {
-				throw new MessageMappingException("Cannot map to message: " + object, e);
-			}
+		return Mono.defer(() -> {
 
-			Object originalReplyChannelHeader = message.getHeaders().getReplyChannel();
-			Object originalErrorChannelHeader = message.getHeaders().getErrorChannel();
+			Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel();
+			Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel();
 
 			MonoReplyChannel replyChan = new MonoReplyChannel();
 
-			Message<?> requestMessage = MutableMessageBuilder.fromMessage(message)
+			Message<?> messageToSend = MutableMessageBuilder.fromMessage(requestMessage)
 					.setReplyChannel(replyChan)
 					.setHeader(this.messagingTemplate.getSendTimeoutHeader(), null)
 					.setHeader(this.messagingTemplate.getReceiveTimeoutHeader(), null)
 					.setErrorChannel(replyChan)
 					.build();
 
-			sendMessageForReactiveFlow(requestChannel, requestMessage);
+			sendMessageForReactiveFlow(requestChannel, messageToSend);
 
 			return buildReplyMono(requestMessage, replyChan.replyMono, error, originalReplyChannelHeader,
-					originalErrorChannelHeader)
-					.onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t));
-		});
+					originalErrorChannelHeader);
+		})
+				.onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t));
 	}
 
 	private void sendMessageForReactiveFlow(MessageChannel requestChannel, Message<?> requestMessage) {
diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/dsl/HttpDslTests.java
@@ -31,9 +31,8 @@
 import java.util.List;
 import java.util.Map;
 
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -45,10 +44,12 @@
 import org.springframework.http.client.ClientHttpRequestFactory;
 import org.springframework.http.client.ClientHttpResponse;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.FixedSubscriberChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.context.IntegrationFlowContext;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.http.multipart.UploadedMultipartFile;
 import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler;
 import org.springframework.integration.security.channel.ChannelSecurityInterceptor;
@@ -68,8 +69,7 @@
 import org.springframework.security.crypto.factory.PasswordEncoderFactories;
 import org.springframework.security.provisioning.InMemoryUserDetailsManager;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
-import org.springframework.test.context.web.WebAppConfiguration;
+import org.springframework.test.context.junit.jupiter.web.SpringJUnitWebConfig;
 import org.springframework.test.web.client.MockMvcClientHttpRequestFactory;
 import org.springframework.test.web.servlet.MockMvc;
 import org.springframework.test.web.servlet.setup.MockMvcBuilders;
@@ -91,8 +91,7 @@
  *
  * @since 5.0
  */
-@RunWith(SpringRunner.class)
-@WebAppConfiguration
+@SpringJUnitWebConfig
 @DirtiesContext
 public class HttpDslTests {
 
@@ -107,7 +106,7 @@ public class HttpDslTests {
 
 	private MockMvc mockMvc;
 
-	@Before
+	@BeforeEach
 	public void setup() {
 		this.mockMvc =
 				MockMvcBuilders.webAppContextSetup(this.wac)
@@ -236,6 +235,37 @@ public void testValidation() throws Exception {
 	}
 
 
+	@Test
+	public void testErrorChannelFlow() throws Exception {
+		IntegrationFlow flow =
+				IntegrationFlows.from(
+						Http.inboundGateway("/errorFlow")
+								.errorChannel(new FixedSubscriberChannel(
+										new AbstractReplyProducingMessageHandler() {
+
+											@Override
+											protected Object handleRequestMessage(Message<?> requestMessage) {
+												return "Error Response";
+											}
+
+										})))
+						.transform((payload) -> {
+							throw new RuntimeException("Error!");
+						})
+						.get();
+
+		IntegrationFlowContext.IntegrationFlowRegistration flowRegistration =
+				this.integrationFlowContext.registration(flow).register();
+
+		this.mockMvc.perform(
+				get("/errorFlow")
+						.with(httpBasic("user", "user")))
+				.andExpect(status().isOk())
+				.andExpect(content().string("Error Response"));
+
+		flowRegistration.destroy();
+	}
+
 	@Configuration
 	@EnableWebSecurity
 	@EnableIntegration
diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java
@@ -28,9 +28,8 @@
 import javax.annotation.Resource;
 
 import org.hamcrest.Matchers;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
 
 import org.springframework.beans.DirectFieldAccessor;
@@ -44,12 +43,14 @@
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.client.reactive.ClientHttpConnector;
+import org.springframework.integration.channel.FixedSubscriberChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.dsl.context.IntegrationFlowContext;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.http.HttpHeaders;
 import org.springframework.integration.http.dsl.Http;
 import org.springframework.integration.support.MessageBuilder;
@@ -76,8 +77,7 @@
 import org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers;
 import org.springframework.security.web.server.SecurityWebFilterChain;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
-import org.springframework.test.context.web.WebAppConfiguration;
+import org.springframework.test.context.junit.jupiter.web.SpringJUnitWebConfig;
 import org.springframework.test.web.reactive.server.HttpHandlerConnector;
 import org.springframework.test.web.reactive.server.WebTestClient;
 import org.springframework.test.web.servlet.MockMvc;
@@ -104,8 +104,7 @@
  *
  * @since 5.0
  */
-@RunWith(SpringRunner.class)
-@WebAppConfiguration
+@SpringJUnitWebConfig
 @DirtiesContext
 public class WebFluxDslTests {
 
@@ -130,7 +129,7 @@ public class WebFluxDslTests {
 
 	private WebTestClient webTestClient;
 
-	@Before
+	@BeforeEach
 	public void setup() {
 		this.mockMvc =
 				MockMvcBuilders.webAppContextSetup(this.wac)
@@ -320,6 +319,39 @@ public void testValidation() {
 		flowRegistration.destroy();
 	}
 
+	@Test
+	public void testErrorChannelFlow() {
+		IntegrationFlow flow =
+				IntegrationFlows.from(
+						WebFlux.inboundGateway("/errorFlow")
+								.errorChannel(new FixedSubscriberChannel(
+										new AbstractReplyProducingMessageHandler() {
+
+											@Override
+											protected Object handleRequestMessage(Message<?> requestMessage) {
+												return "Error Response";
+											}
+
+										})))
+						.channel(MessageChannels.flux())
+						.transform((payload) -> {
+							throw new RuntimeException("Error!");
+						})
+						.get();
+
+		IntegrationFlowContext.IntegrationFlowRegistration flowRegistration =
+				this.integrationFlowContext.registration(flow).register();
+
+		this.webTestClient.get().uri("/errorFlow")
+				.headers(headers -> headers.setBasicAuth("guest", "guest"))
+				.exchange()
+				.expectStatus()
+				.isOk()
+				.expectBody(String.class)
+				.isEqualTo("Error Response");
+
+		flowRegistration.destroy();
+	}
 
 	@Configuration
 	@EnableWebFlux
@@ -450,7 +482,7 @@ public IntegrationFlow sseFlow() {
 					.from(WebFlux.inboundGateway("/sse")
 							.requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE))
 							.mappedResponseHeaders("*"))
-					.enrichHeaders(Collections.singletonMap("aHeader", new String[] { "foo", "bar", "baz" }))
+					.enrichHeaders(Collections.singletonMap("aHeader", new String[]{"foo", "bar", "baz"}))
 					.handle((p, h) -> Flux.fromArray(h.get("aHeader", String[].class)))
 					.get();
 		}
