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

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

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 {
 					<Message<?>>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<? super Message<?>>) subscriber);
@@ -125,7 +126,7 @@ public final class MessageChannelReactiveUtils {
 
 			};
 
-			Mono.<Message<?>>delayMillis(100)
+			Mono.<Message<?>>delay(Duration.ofMillis(100))
 					.repeat()
 					.concatMap(value -> Flux.fromIterable(() -> messageIterator))
 					.subscribe((Subscriber<? super Message<?>>) subscriber);

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<?>, Message<?>> processor;
 
+	private final Flux<Message<?>> flux;
+
 	private final BlockingSink<Message<?>> sink;
 
 	private volatile boolean upstreamSubscribed;
@@ -57,6 +58,7 @@ public class ReactiveChannel extends AbstractMessageChannel
 	public ReactiveChannel(Processor<Message<?>, 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<? super Message<?>> subscriber) {
 		this.subscribers.add(subscriber);
-		this.processor.subscribe(new Operators.SubscriberAdapter<Message<?>, 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<Message<?>> publisher) {
-		publisher.subscribe(new Operators.SubscriberAdapter<Message<?>, 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);
 	}
 
 }

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<?>, Message<?>> subscriber;
+	private final Publisher<Message<Object>> publisher;
+
+	private final Subscriber<Message<?>> subscriber;
 
 	private final Lifecycle lifecycleDelegate;
 
-	private volatile Publisher<Message<?>> 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<Message<?>> subscriber) {
+	public ReactiveConsumer(MessageChannel inputChannel, final Subscriber<Message<?>> 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<Message<?>>) messagePublisher;
+		this.publisher = MessageChannelReactiveUtils.toPublisher(inputChannel);
 
-		this.subscriber = new Operators.SubscriberAdapter<Message<?>, 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<Message<?>>() {
+
+			private final Subscriber<Message<?>> 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<Message<?>>, Receiver, Disposable, Trackable, Lifecycle {
+			implements Subscriber<Message<?>>, Disposable, Lifecycle {
 
 		private final Consumer<Message<?>> 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;
 		}
 

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");