Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4541: Fix Reactive MessagingGateway Errors

Browse Source 
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.
This commit is contained in:
Gary Russell
2018-10-08 16:20:08 -04:00
committed by Artem Bilan
parent 9ae0eb2dbc
commit e7bc060e55
2 changed files with 52 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -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));
});
}

35
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java
View File

@@ -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