Upgrade to Reactor 2020.0.0-RC1

* Handle `Emission.FAIL_ZERO_SUBSCRIBER` in the
`FluxMessageChannel` and `IntegrationReactiveUtils`

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'

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<? super Message<?>> subscriber) {
 		this.processor
 				.doFinally((s) -> this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams()))
+				.share()
 				.subscribe(subscriber);
 		this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams());
 	}

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 " +