From 481d5eb2a56abe92466c784fb24809bbfedc0a9f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 11 Sep 2020 12:35:14 -0400 Subject: [PATCH] Fix new Sonar smells * Use `tryEmitNext` on Reactor `Sink` since `emitNext` is deprecated * Add `MessageDeliveryException` emission when `send()` returns `false` in the `FluxMessageChannel` for `subscribeTo` provided `Publisher` --- .../integration/channel/FluxMessageChannel.java | 15 ++++++++++----- .../util/IntegrationReactiveUtils.java | 2 +- .../reactive/ReactiveStreamsConsumerTests.java | 2 +- .../RSocketOutboundGatewayIntegrationTests.java | 2 +- 4 files changed, 13 insertions(+), 8 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 26b754eecd..9a8118fc6a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -23,6 +23,7 @@ import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; import org.springframework.util.Assert; import reactor.core.Disposable; @@ -71,7 +72,8 @@ public class FluxMessageChannel extends AbstractMessageChannel long parkTimeout = 10; // NOSONAR long parkTimeoutNs = TimeUnit.MILLISECONDS.toNanos(parkTimeout); while (this.active && !tryEmitMessage(message)) { - if (timeout >= 0 && (remainingTime -= parkTimeout) <= 0) { + remainingTime -= parkTimeout; + if (timeout >= 0 && remainingTime <= 0) { return false; } LockSupport.parkNanos(parkTimeoutNs); @@ -95,9 +97,9 @@ public class FluxMessageChannel extends AbstractMessageChannel @Override public void subscribe(Subscriber> subscriber) { this.processor - .doFinally((s) -> this.subscribedSignal.emitNext(this.processor.hasDownstreams())) + .doFinally((s) -> this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams())) .subscribe(subscriber); - this.subscribedSignal.emitNext(this.processor.hasDownstreams()); + this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams()); } @Override @@ -108,7 +110,10 @@ public class FluxMessageChannel extends AbstractMessageChannel .publishOn(Schedulers.boundedElastic()) .doOnNext((message) -> { try { - send(message); + if (!send(message)) { + throw new MessageDeliveryException(message, + "Failed to send message to channel '" + this); + } } catch (Exception ex) { logger.warn("Error during processing event: " + message, ex); @@ -120,7 +125,7 @@ public class FluxMessageChannel extends AbstractMessageChannel @Override public void destroy() { this.active = false; - this.subscribedSignal.emitNext(false); + this.subscribedSignal.tryEmitNext(false); this.upstreamSubscriptions.dispose(); this.processor.onComplete(); super.destroy(); 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 8bee8807bc..9d5ee1bc0c 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 @@ -100,7 +100,7 @@ public final class IntegrationReactiveUtils { * - a {@link org.springframework.integration.channel.FluxMessageChannel} * is returned as is because it is already a {@link Publisher}; * - a {@link SubscribableChannel} is subscribed with a {@link MessageHandler} - * for the {@link Sinks.Many#emitNext(Object)} which is returned from this method; + * for the {@link Sinks.Many#tryEmitNext(Object)} which is returned from this method; * - a {@link PollableChannel} is wrapped into a {@link MessageSource} lambda and reuses * {@link #messageSourceToFlux(MessageSource)}. * @param messageChannel the {@link MessageChannel} to adapt. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java index d6311bd6e0..109bddd8a4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java @@ -301,7 +301,7 @@ public class ReactiveStreamsConsumerTests { ReactiveMessageHandler messageHandler = m -> { - sink.emitNext(m); + sink.tryEmitNext(m); return Mono.empty(); }; diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java index 9d59498a15..05a8c201bd 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java @@ -544,7 +544,7 @@ public class RSocketOutboundGatewayIntegrationTests { @MessageMapping("receive") void receive(String payload) { - this.fireForgetPayloads.emitNext(payload); + this.fireForgetPayloads.tryEmitNext(payload); } @MessageMapping("echo")