From 499e1167d1619c2d93b4cf07e61e4fa0ed47ec18 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 3 Apr 2017 14:00:32 -0400 Subject: [PATCH] Upgrade to Reactor-3.0.6 According to some rework and deprecations in the latest Reactor, fix Reactor-based components to follow with the latest paradigmas * Use `Flux` wrapper in the `ReactiveChannel` to track cancellation instead of deprecated `Operators.SubscriberAdapter` * Do similar `Flux.from(publisher)` in the `doSubscribeTo()` to avoid subscribers delegation and, therefore, unnecessary call stack * Rework `ReactiveConsumer` to wrap to a new `BaseSubscriber` on each `ReactiveConsumer.start()`. We need this wrapping to prevent subscriber cancellation from the upstream `Publisher` on error. We can't rely here on the `Flux.retry()` because upstream `DirectProcessor` is marked as terminated on any error Apply Reactor-3.0.6.RELEASE --- build.gradle | 2 +- .../channel/MessageChannelReactiveUtils.java | 5 +- .../integration/channel/ReactiveChannel.java | 43 ++++------ .../endpoint/ReactiveConsumer.java | 83 ++++++++++--------- .../reactive/ReactiveConsumerTests.java | 5 ++ 5 files changed, 66 insertions(+), 72 deletions(-) diff --git a/build.gradle b/build.gradle index 61c2a58ddf..584a06a034 100644 --- a/build.gradle +++ b/build.gradle @@ -125,7 +125,7 @@ subprojects { subproject -> pahoMqttClientVersion = '1.0.2' postgresVersion = '9.1-901-1.jdbc4' reactorNettyVersion = '0.6.2.RELEASE' - reactorVersion = '3.0.5.RELEASE' + reactorVersion = '3.0.6.RELEASE' romeToolsVersion = '1.7.0' servletApiVersion = '3.1.0' slf4jVersion = "1.7.21" diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java index 506c64bfa7..d84ce87591 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java @@ -16,6 +16,7 @@ package org.springframework.integration.channel; +import java.time.Duration; import java.util.Iterator; import org.reactivestreams.Publisher; @@ -85,7 +86,7 @@ public final class MessageChannelReactiveUtils { >create(emitter -> { MessageHandler messageHandler = emitter::next; this.channel.subscribe(messageHandler); - emitter.setCancellation(() -> this.channel.unsubscribe(messageHandler)); + emitter.onCancel(() -> this.channel.unsubscribe(messageHandler)); }, FluxSink.OverflowStrategy.IGNORE) .subscribe((Subscriber>) subscriber); @@ -125,7 +126,7 @@ public final class MessageChannelReactiveUtils { }; - Mono.>delayMillis(100) + Mono.>delay(Duration.ofMillis(100)) .repeat() .concatMap(value -> Flux.fromIterable(() -> messageIterator)) .subscribe((Subscriber>) subscriber); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java index 5eace718db..52e19bce74 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java @@ -23,14 +23,13 @@ import java.util.concurrent.CopyOnWriteArrayList; import org.reactivestreams.Processor; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; import org.springframework.messaging.Message; import org.springframework.util.Assert; import reactor.core.publisher.BlockingSink; import reactor.core.publisher.DirectProcessor; -import reactor.core.publisher.Operators; +import reactor.core.publisher.Flux; /** * @author Artem Bilan @@ -46,6 +45,8 @@ public class ReactiveChannel extends AbstractMessageChannel private final Processor, Message> processor; + private final Flux> flux; + private final BlockingSink> sink; private volatile boolean upstreamSubscribed; @@ -57,6 +58,7 @@ public class ReactiveChannel extends AbstractMessageChannel public ReactiveChannel(Processor, Message> processor) { Assert.notNull(processor, "'processor' must not be null"); this.processor = processor; + this.flux = Flux.from(processor); this.sink = BlockingSink.create(this.processor); } @@ -68,15 +70,9 @@ public class ReactiveChannel extends AbstractMessageChannel @Override public void subscribe(Subscriber> subscriber) { this.subscribers.add(subscriber); - this.processor.subscribe(new Operators.SubscriberAdapter, Message>(subscriber) { - @Override - protected void doCancel() { - super.doCancel(); - ReactiveChannel.this.subscribers.remove(subscriber); - } - - }); + this.flux.doOnCancel(() -> ReactiveChannel.this.subscribers.remove(subscriber)) + .subscribe(subscriber); if (!this.upstreamSubscribed) { this.publishers.forEach(this::doSubscribeTo); @@ -92,24 +88,15 @@ public class ReactiveChannel extends AbstractMessageChannel } private void doSubscribeTo(Publisher> publisher) { - publisher.subscribe(new Operators.SubscriberAdapter, Message>(this.processor) { - - @Override - protected void doOnSubscribe(Subscription subscription) { - super.doOnSubscribe(subscription); - ReactiveChannel.this.upstreamSubscribed = true; - } - - @Override - protected void doComplete() { - super.doComplete(); - ReactiveChannel.this.publishers.remove(publisher); - if (ReactiveChannel.this.publishers.isEmpty()) { - ReactiveChannel.this.upstreamSubscribed = false; - } - } - - }); + Flux.from(publisher) + .doOnSubscribe(s -> ReactiveChannel.this.upstreamSubscribed = true) + .doOnComplete(() -> { + ReactiveChannel.this.publishers.remove(publisher); + if (ReactiveChannel.this.publishers.isEmpty()) { + ReactiveChannel.this.upstreamSubscribed = false; + } + }) + .subscribe(this.processor); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java index d927d3bb7b..1f00ebe4df 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java @@ -34,8 +34,7 @@ import org.springframework.util.ErrorHandler; import reactor.core.Disposable; import reactor.core.Exceptions; -import reactor.core.Receiver; -import reactor.core.Trackable; +import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Operators; @@ -45,14 +44,16 @@ import reactor.core.publisher.Operators; */ public class ReactiveConsumer extends AbstractEndpoint { - private final Operators.SubscriberAdapter, Message> subscriber; + private final Publisher> publisher; + + private final Subscriber> subscriber; private final Lifecycle lifecycleDelegate; - private volatile Publisher> publisher; - private ErrorHandler errorHandler; + private volatile Subscription subscription; + @SuppressWarnings("unchecked") public ReactiveConsumer(MessageChannel inputChannel, MessageHandler messageHandler) { this(inputChannel, @@ -61,28 +62,14 @@ public class ReactiveConsumer extends AbstractEndpoint { : new MessageHandlerSubscriber(messageHandler)); } - @SuppressWarnings("unchecked") - public ReactiveConsumer(MessageChannel inputChannel, Subscriber> subscriber) { + public ReactiveConsumer(MessageChannel inputChannel, final Subscriber> subscriber) { Assert.notNull(inputChannel, "'inputChannel' must not be null"); Assert.notNull(subscriber, "'subscriber' must not be null"); - Publisher messagePublisher = MessageChannelReactiveUtils.toPublisher(inputChannel); - this.publisher = (Publisher>) messagePublisher; + this.publisher = MessageChannelReactiveUtils.toPublisher(inputChannel); - this.subscriber = new Operators.SubscriberAdapter, Message>(subscriber) { + this.subscriber = subscriber; - @Override - protected void doNext(Message message) { - try { - super.doNext(message); - } - catch (Exception e) { - ReactiveConsumer.this.errorHandler.handleError(e); - doOnSubscriberError(e); - } - } - - }; this.lifecycleDelegate = subscriber instanceof Lifecycle ? (Lifecycle) subscriber : null; } @@ -104,12 +91,41 @@ public class ReactiveConsumer extends AbstractEndpoint { if (this.lifecycleDelegate != null) { this.lifecycleDelegate.start(); } - this.publisher.subscribe(this.subscriber); + this.publisher.subscribe(new BaseSubscriber>() { + + private final Subscriber> delegate = ReactiveConsumer.this.subscriber; + + public void hookOnSubscribe(Subscription s) { + this.delegate.onSubscribe(s); + ReactiveConsumer.this.subscription = s; + } + + public void hookOnNext(Message message) { + try { + this.delegate.onNext(message); + } + catch (Exception e) { + ReactiveConsumer.this.errorHandler.handleError(e); + hookOnError(e); + } + } + + public void hookOnError(Throwable t) { + this.delegate.onError(t); + } + + public void hookOnComplete() { + this.delegate.onComplete(); + } + + }); } @Override protected void doStop() { - this.subscriber.cancel(); + if (this.subscription != null) { + this.subscription.cancel(); + } if (this.lifecycleDelegate != null) { this.lifecycleDelegate.stop(); } @@ -117,7 +133,7 @@ public class ReactiveConsumer extends AbstractEndpoint { private static final class MessageHandlerSubscriber - implements Subscriber>, Receiver, Disposable, Trackable, Lifecycle { + implements Subscriber>, Disposable, Lifecycle { private final Consumer> consumer; @@ -158,11 +174,6 @@ public class ReactiveConsumer extends AbstractEndpoint { } } - @Override - public Object upstream() { - return this.subscription; - } - @Override public void dispose() { Subscription s = this.subscription; @@ -173,17 +184,7 @@ public class ReactiveConsumer extends AbstractEndpoint { } @Override - public long getCapacity() { - return Long.MAX_VALUE; - } - - @Override - public boolean isStarted() { - return this.subscription != null; - } - - @Override - public boolean isTerminated() { + public boolean isDisposed() { return false; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java index 2cddf87560..f6512b28f5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java @@ -26,6 +26,7 @@ import static org.mockito.BDDMockito.willAnswer; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import java.util.LinkedList; @@ -59,6 +60,7 @@ import reactor.core.publisher.EmitterProcessor; /** * @author Artem Bilan + * * @since 5.0 */ public class ReactiveConsumerTests { @@ -199,6 +201,9 @@ public class ReactiveConsumerTests { reactiveConsumer.start(); + verify(testSubscriber, times(2)).onSubscribe(subscriptionArgumentCaptor.capture()); + subscription = subscriptionArgumentCaptor.getValue(); + subscription.request(2); Message testMessage2 = new GenericMessage<>("test2");