From a2e70d2c7826e5dea2f6b5759359b382c1686799 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 19 Aug 2020 10:20:44 -0400 Subject: [PATCH] Move back to SNAPSHOTs; fix for Reactor changes --- build.gradle | 10 +++++----- .../integration/channel/FluxMessageChannel.java | 2 +- .../integration/gateway/MessagingGatewaySupport.java | 2 +- .../integration/util/IntegrationReactiveUtils.java | 2 +- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/build.gradle b/build.gradle index a462b89e86..db962d8162 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-M2' + reactorVersion = '2020.0.0-SNAPSHOT' resilience4jVersion = '1.5.0' romeToolsVersion = '1.12.2' rsocketVersion = '1.1.0-M1' @@ -98,12 +98,12 @@ ext { servletApiVersion = '4.0.1' smackVersion = '4.3.4' soapVersion = '1.4.0' - springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.0-M2' - springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2020.0.0-M2' - springKafkaVersion = '2.6.0-M1' + springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.0-SNAPSHOT' + springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2020.0.0-SNAPSHOT' + springKafkaVersion = '2.6.0-SNAPSHOT' springRetryVersion = '1.3.0' springSecurityVersion = project.hasProperty('springSecurityVersion') ? project.springSecurityVersion : '5.4.0-RC1' - springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.3.0-M2' + springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.3.0-SNAPSHOT' springWsVersion = '3.0.9.RELEASE' tomcatVersion = "9.0.37" xstreamVersion = '1.4.12' 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 cb5871e55e..7f6eaa3e1b 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 @@ -59,7 +59,7 @@ public class FluxMessageChannel extends AbstractMessageChannel protected boolean doSend(Message message, long timeout) { Assert.state(this.processor.hasDownstreams(), () -> "The [" + this + "] doesn't have subscribers to accept messages"); - return this.sink.emitNext(message).hasEmitted(); + return this.sink.tryEmitNext(message).hasSucceeded(); } @Override 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 650ac26fcf..10043075db 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 @@ -894,7 +894,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint @Override public boolean send(Message message, long timeout) { - return Boolean.TRUE.equals(this.replyMono.emitValue(message).hasEmitted()); + return this.replyMono.tryEmitValue(message).hasSucceeded(); } @Override 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 7ef90c7001..61940a13bf 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 @@ -129,7 +129,7 @@ public final class IntegrationReactiveUtils { Sinks.Many> sink = Sinks.many().multicast().onBackpressureBuffer(1); @SuppressWarnings("unchecked") MessageHandler messageHandler = (message) -> { - while (!sink.emitNext((Message) message).hasEmitted()) { + while (!sink.tryEmitNext((Message) message).hasSucceeded()) { LockSupport.parkNanos(100); // NOSONAR } };