From 19b59bbde87068e8a377fc95f588c33b779497ad Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 14 Sep 2020 11:59:20 -0400 Subject: [PATCH] Upgrade to Reactor 2020.0.0-RC1 * Handle `Emission.FAIL_ZERO_SUBSCRIBER` in the `FluxMessageChannel` and `IntegrationReactiveUtils` --- build.gradle | 2 +- .../integration/channel/FluxMessageChannel.java | 3 +++ .../integration/util/IntegrationReactiveUtils.java | 3 +++ 3 files changed, 7 insertions(+), 1 deletion(-) diff --git a/build.gradle b/build.gradle index d2dd7e4e66..0af8cf3eb3 100644 --- a/build.gradle +++ b/build.gradle @@ -90,7 +90,7 @@ ext { pahoMqttClientVersion = '1.2.4' postgresVersion = '42.2.14' r2dbch2Version='0.8.4.RELEASE' - reactorVersion = '2020.0.0-SNAPSHOT' + reactorVersion = '2020.0.0-RC1' resilience4jVersion = '1.5.0' romeToolsVersion = '1.12.2' rsocketVersion = '1.1.0-SNAPSHOT' 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 41a4c9f318..012498636e 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 @@ -88,6 +88,8 @@ public class FluxMessageChannel extends AbstractMessageChannel case FAIL_NON_SERIALIZED: case FAIL_OVERFLOW: return false; + case FAIL_ZERO_SUBSCRIBER: + throw new IllegalStateException("The [" + this + "] doesn't have subscribers to accept messages"); case FAIL_TERMINATED: case FAIL_CANCELLED: throw new IllegalStateException("Cannot emit messages into the cancelled or terminated sink: " @@ -101,6 +103,7 @@ public class FluxMessageChannel extends AbstractMessageChannel public void subscribe(Subscriber> subscriber) { this.processor .doFinally((s) -> this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams())) + .share() .subscribe(subscriber); this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams()); } 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 ec20cd24bc..ff670fc0f3 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 @@ -135,6 +135,9 @@ public final class IntegrationReactiveUtils { case FAIL_OVERFLOW: LockSupport.parkNanos(1000); // NOSONAR break; + case FAIL_ZERO_SUBSCRIBER: + throw new IllegalStateException("The [" + sink + + "] doesn't have subscribers to accept messages"); case FAIL_TERMINATED: case FAIL_CANCELLED: throw new IllegalStateException("Cannot emit messages into the cancelled " +