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

Fix new Sonar smells

Browse Source 
* Use `tryEmitNext` on Reactor `Sink` since `emitNext` is deprecated
* Add `MessageDeliveryException` emission when `send()` returns `false`
in the `FluxMessageChannel` for `subscribeTo` provided `Publisher`
This commit is contained in:
Artem Bilan
2020-09-11 12:35:14 -04:00
parent 383af8ceb9
commit 481d5eb2a5
4 changed files with 13 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<? super Message<?>> 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();

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

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

2
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java
View File

@@ -301,7 +301,7 @@ public class ReactiveStreamsConsumerTests {
ReactiveMessageHandler messageHandler =
m -> {
sink.emitNext(m);
sink.tryEmitNext(m);
return Mono.empty();
};

2
spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java
View File

@@ -544,7 +544,7 @@ public class RSocketOutboundGatewayIntegrationTests {
@MessageMapping("receive")
void receive(String payload) {
this.fireForgetPayloads.emitNext(payload);
this.fireForgetPayloads.tryEmitNext(payload);
}
@MessageMapping("echo")