From e7bc060e55b876fe0ffe4ef54fb6b80b4d4e739f Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Mon, 8 Oct 2018 16:20:08 -0400
Subject: [PATCH] INT-4541: Fix Reactive MessagingGateway Errors

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

Add test case to reproduce.

The gateway correctly sets up the `errorChannel` header so that a downstream
`MPEH` will send exceptions back to the gateway, but the `map()` function
did not check for an error message.

Check for an error message and throw the payload so that the `onErrorResume`
will route to the error channel.
---
 .../gateway/MessagingGatewaySupport.java      | 23 ++++++++----
 .../webflux/dsl/WebFluxDslTests.java          | 35 +++++++++++++++++++
 2 files changed, 52 insertions(+), 6 deletions(-)

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
index 5d1c941f73..f53e2eb24b 100644
--- 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
@@ -628,12 +628,23 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 							this.messageCount.incrementAndGet();
 						}
 					})
-					.<Message<?>>map(replyMessage ->
-							MessageBuilder.fromMessage(replyMessage)
-									.setHeader(MessageHeaders.REPLY_CHANNEL, originalReplyChannelHeader)
-									.setHeader(MessageHeaders.ERROR_CHANNEL, originalErrorChannelHeader)
-									.build())
-
+					.<Message<?>>map(replyMessage -> {
+							if (!error && replyMessage instanceof ErrorMessage) {
+								ErrorMessage em = (ErrorMessage) replyMessage;
+								if (em.getPayload() instanceof MessagingException) {
+									throw (MessagingException) em.getPayload();
+								}
+								else {
+									throw new MessagingException(requestMessage, em.getPayload());
+								}
+							}
+							else {
+								return MessageBuilder.fromMessage(replyMessage)
+										.setHeader(MessageHeaders.REPLY_CHANNEL, originalReplyChannelHeader)
+										.setHeader(MessageHeaders.ERROR_CHANNEL, originalErrorChannelHeader)
+										.build();
+							}
+					})
 					.onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t));
 		});
 	}
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
index 52c95bc542..c30f6e7e0b 100644
--- 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
@@ -49,6 +49,7 @@ 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.http.HttpHeaders;
 import org.springframework.integration.http.dsl.Http;
@@ -137,6 +138,7 @@ public class WebFluxDslTests {
 				WebTestClient.bindToApplicationContext(this.wac)
 						.apply(SecurityMockServerConfigurers.springSecurity())
 						.configureClient()
+//						.responseTimeout(Duration.ofSeconds(600))
 						.build();
 	}
 
@@ -232,6 +234,15 @@ public class WebFluxDslTests {
 
 	}
 
+	@Test
+	public void testHttpReactivePostWithError() {
+		this.webTestClient.post().uri("/reactivePostErrors")
+				.headers(headers -> headers.setBasicAuth("guest", "guest"))
+				.body(Mono.just("foo\nbar\nbaz"), String.class)
+				.exchange()
+				.expectStatus().isEqualTo(HttpStatus.BAD_GATEWAY);
+	}
+
 	@Test
 	public void testSse() {
 		Flux<String> responseBody =
@@ -363,6 +374,30 @@ public class WebFluxDslTests {
 					.get();
 		}
 
+		@Bean
+		public IntegrationFlow httpReactiveInboundGatewayFlowWithErrors() {
+			return IntegrationFlows
+					.from(WebFlux.inboundGateway("/reactivePostErrors")
+							.requestMapping(m -> m.methods(HttpMethod.POST))
+							.requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class))
+							.statusCodeFunction(e ->
+									HttpMethod.POST.equals(e.getMethod())
+											? HttpStatus.ACCEPTED
+											: HttpStatus.BAD_REQUEST)
+							.errorChannel(errorFlow().getInputChannel()))
+					.channel(MessageChannels.flux())
+					.handle((p, h) -> {
+						throw new RuntimeException("errorTest");
+					})
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow errorFlow() {
+			return f -> f
+					.enrichHeaders(h -> h.header(HttpHeaders.STATUS_CODE, HttpStatus.BAD_GATEWAY));
+		}
+
 		@Bean
 		public IntegrationFlow sseFlow() {
 			return IntegrationFlows