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 0e78dee1b7..c37dc6786b 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 @@ -623,12 +623,23 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint this.messageCount.incrementAndGet(); } }) - .>map(replyMessage -> - MessageBuilder.fromMessage(replyMessage) - .setHeader(MessageHeaders.REPLY_CHANNEL, originalReplyChannelHeader) - .setHeader(MessageHeaders.ERROR_CHANNEL, originalErrorChannelHeader) - .build()) - + .>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 df6118bda5..aab48d541d 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 @@ -51,6 +51,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.channel.MessageChannels; import org.springframework.integration.http.HttpHeaders; import org.springframework.integration.http.dsl.Http; import org.springframework.integration.support.MessageBuilder; @@ -231,6 +232,15 @@ public class WebFluxDslTests { } + @Test + public void testHttpReactivePostWithError() { + this.webTestClient.post().uri("/reactivePostErrors") + .attributes(basicAuthenticationCredentials("guest", "guest")) + .body(Mono.just("foo\nbar\nbaz"), String.class) + .exchange() + .expectStatus().isEqualTo(HttpStatus.BAD_GATEWAY); + } + @Test public void testSse() { Flux responseBody = @@ -332,6 +342,26 @@ 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)) + .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