From 619c5bfa13d0ef1f013d9b33767cd5dd59463bc0 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 13 May 2020 14:13:40 -0400 Subject: [PATCH] GH-3276: reactive inbound: Fix `onErrorResume` Fixes https://github.com/spring-projects/spring-integration/issues/3276 The `onErrorResume` for the `MessagingGatewaySupport.doSendAndReceiveMessageReactive()` was in wrong place: only for the `buildReplyMono` which works only when an outbound flow is fully based on reactive channels. With a regular direct channel we can get an exception from the `sendMessageForReactiveFlow` which is not covered with the mentioned `onErrorResume` for the error handling on the configured `errorChannel` Cherry-pick to `5.2.x & 5.1.x` --- .../integration/gateway/MessagingGatewaySupport.java | 6 +++--- 1 file changed, 3 insertions(+), 3 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 43d70a4570..a3f2a68101 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 @@ -636,7 +636,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint sendMessageForReactiveFlow(requestChannel, requestMessage); return buildReplyMono(requestMessage, replyChan.replyMono, error, originalReplyChannelHeader, - originalErrorChannelHeader); + originalErrorChannelHeader) + .onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t)); }); } @@ -686,8 +687,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint .setHeader(MessageHeaders.ERROR_CHANNEL, originalErrorChannelHeader) .build(); } - }) - .onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t)); + }); } private Mono> handleSendError(Message requestMessage, Throwable exception) {