From c94eaf7b5247940c698b1cd20701216fd8db8ffc Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 23 Sep 2020 15:33:17 -0400 Subject: [PATCH] Fix `IntegrationReactiveUtils` The `Mono.doOnSuccess()` is always called for completed `MonoSink` even if the value is `null` * Fix `IntegrationReactiveUtils.messageSourceToFlux()` to check a message for `null` before calling `AckUtils.autoAck()` * Add an `logger.error()` message when `doOnError()` is invoked for that `Mono` **Cherry-pick to 5.3.x** --- .../integration/util/IntegrationReactiveUtils.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java index ff670fc0f3..70cb26ad2f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java @@ -19,6 +19,8 @@ package org.springframework.integration.util; import java.time.Duration; import java.util.concurrent.locks.LockSupport; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; import org.springframework.integration.StaticMessageHeaderAccessor; @@ -45,6 +47,8 @@ import reactor.core.scheduler.Schedulers; */ public final class IntegrationReactiveUtils { + private static final Log logger = LogFactory.getLog(IntegrationReactiveUtils.class); + /** * The subscriber context entry for {@link Flux#delayElements} * from the {@link Mono#repeatWhenEmpty(java.util.function.Function)}. @@ -76,14 +80,18 @@ public final class IntegrationReactiveUtils { return Mono. >create(monoSink -> monoSink.onRequest(value -> monoSink.success(messageSource.receive()))) - .doOnSuccess((message) -> - AckUtils.autoAck(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message))) + .doOnSuccess((message) -> { + if (message != null) { + AckUtils.autoAck(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message)); + } + }) .doOnError(MessagingException.class, (ex) -> { Message failedMessage = ex.getFailedMessage(); if (failedMessage != null) { AckUtils.autoNack(StaticMessageHeaderAccessor.getAcknowledgmentCallback(failedMessage)); } + logger.error("Error from Flux for : " + messageSource, ex); }) .subscribeOn(Schedulers.boundedElastic()) .repeatWhenEmpty((repeat) ->