Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move back to SNAPSHOTs; fix for Reactor changes

Browse Source
This commit is contained in:
Artem Bilan
2020-08-19 10:20:44 -04:00
parent f45206a013
commit a2e70d2c78
4 changed files with 8 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
build.gradle
View File

@@ -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'

2
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -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

2
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -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

2
spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java
View File

@@ -129,7 +129,7 @@ public final class IntegrationReactiveUtils {
Sinks.Many<Message<T>> sink = Sinks.many().multicast().onBackpressureBuffer(1);
@SuppressWarnings("unchecked")
MessageHandler messageHandler = (message) -> {
while (!sink.emitNext((Message<T>) message).hasEmitted()) {
while (!sink.tryEmitNext((Message<T>) message).hasSucceeded()) {
LockSupport.parkNanos(100); // NOSONAR
}
};