Upgrade to Reactor 3.1

Polishing

- remove catch from test
- update netty for stomp

Add reactiveErrorHandler to AbstractMessageHandler

spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java

@@ -20,19 +20,20 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 
-import org.reactivestreams.Processor;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
-import reactor.core.publisher.BlockingSink;
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxProcessor;
+import reactor.core.publisher.FluxSink;
 
 /**
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  */
@@ -43,11 +44,11 @@ public class ReactiveChannel extends AbstractMessageChannel
 
 	private final List<Publisher<Message<?>>> publishers = new CopyOnWriteArrayList<>();
 
-	private final Processor<Message<?>, Message<?>> processor;
+	private final FluxProcessor<Message<?>, Message<?>> processor;
 
 	private final Flux<Message<?>> flux;
 
-	private final BlockingSink<Message<?>> sink;
+	private final FluxSink<Message<?>> sink;
 
 	private volatile boolean upstreamSubscribed;
 
@@ -55,16 +56,17 @@ public class ReactiveChannel extends AbstractMessageChannel
 		this(DirectProcessor.create());
 	}
 
-	public ReactiveChannel(Processor<Message<?>, Message<?>> processor) {
+	public ReactiveChannel(FluxProcessor<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);
+		this.sink = processor.sink();
 	}
 
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
-		return this.sink.submit(message, timeout) > -1;
+		this.sink.next(message);
+		return true;
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java

@@ -19,8 +19,6 @@ package org.springframework.integration.dsl;
 import java.util.Queue;
 import java.util.concurrent.Executor;
 
-import org.reactivestreams.Processor;
-
 import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.ExecutorChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
@@ -33,6 +31,8 @@ import org.springframework.integration.store.ChannelMessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.messaging.Message;
 
+import reactor.core.publisher.FluxProcessor;
+
 /**
  * @author Artem Bilan
  * @author Gary Russell
@@ -140,11 +140,11 @@ public class Channels {
 		return MessageChannels.reactive(id);
 	}
 
-	public ReactiveChannelSpec reactive(Processor<Message<?>, Message<?>> processor) {
+	public ReactiveChannelSpec reactive(FluxProcessor<Message<?>, Message<?>> processor) {
 		return MessageChannels.reactive(processor);
 	}
 
-	public ReactiveChannelSpec reactive(String id, Processor<Message<?>, Message<?>> processor) {
+	public ReactiveChannelSpec reactive(String id, FluxProcessor<Message<?>, Message<?>> processor) {
 		return MessageChannels.reactive(id, processor);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java

@@ -19,14 +19,15 @@ package org.springframework.integration.dsl.channel;
 import java.util.Queue;
 import java.util.concurrent.Executor;
 
-import org.reactivestreams.Processor;
-
 import org.springframework.integration.store.ChannelMessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.messaging.Message;
 
+import reactor.core.publisher.FluxProcessor;
+
 /**
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  */
@@ -133,11 +134,11 @@ public final class MessageChannels {
 		return reactive().id(id);
 	}
 
-	public static ReactiveChannelSpec reactive(Processor<Message<?>, Message<?>> processor) {
+	public static ReactiveChannelSpec reactive(FluxProcessor<Message<?>, Message<?>> processor) {
 		return new ReactiveChannelSpec(processor);
 	}
 
-	public static ReactiveChannelSpec reactive(String id, Processor<Message<?>, Message<?>> processor) {
+	public static ReactiveChannelSpec reactive(String id, FluxProcessor<Message<?>, Message<?>> processor) {
 		return reactive(processor).id(id);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java

@@ -16,13 +16,14 @@
 
 package org.springframework.integration.dsl.channel;
 
-import org.reactivestreams.Processor;
-
 import org.springframework.integration.channel.ReactiveChannel;
 import org.springframework.messaging.Message;
 
+import reactor.core.publisher.FluxProcessor;
+
 /**
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  */
@@ -32,7 +33,7 @@ public class ReactiveChannelSpec extends MessageChannelSpec<ReactiveChannelSpec,
 		this.channel = new ReactiveChannel();
 	}
 
-	ReactiveChannelSpec(Processor<Message<?>, Message<?>> processor) {
+	ReactiveChannelSpec(FluxProcessor<Message<?>, Message<?>> processor) {
 		this.channel = new ReactiveChannel(processor);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java

@@ -20,6 +20,7 @@ import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
 import org.springframework.core.Ordered;
+import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.context.Orderable;
 import org.springframework.integration.history.MessageHistory;
@@ -36,6 +37,7 @@ import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
+import org.springframework.util.ErrorHandler;
 
 import reactor.core.publisher.Operators;
 
@@ -70,6 +72,8 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 
 	private volatile boolean loggingEnabled = true;
 
+	private ErrorHandler reactiveErrorHandler;
+
 	@Override
 	public boolean isLoggingEnabled() {
 		return this.loggingEnabled;
@@ -100,6 +104,16 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		this.shouldTrack = shouldTrack;
 	}
 
+	/**
+	 * Set the error handler to use when an exception occurs when this handler
+	 * is invoked as a reactive {@link Subscriber}.
+	 * @param reactiveErrorHandler the error handler.
+	 * @since 5.0
+	 */
+	public void setReactiveErrorHandler(ErrorHandler reactiveErrorHandler) {
+		this.reactiveErrorHandler = reactiveErrorHandler;
+	}
+
 	@Override
 	public void configureMetrics(AbstractMessageHandlerMetrics metrics) {
 		Assert.notNull(metrics, "'metrics' must not be null");
@@ -111,6 +125,14 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		if (this.statsEnabled) {
 			this.handlerMetrics.setFullStatsEnabled(true);
 		}
+		if (this.reactiveErrorHandler == null) {
+			if (getBeanFactory() != null) {
+				this.reactiveErrorHandler = new MessagePublishingErrorHandler(getChannelResolver());
+			}
+			else {
+				this.reactiveErrorHandler = new MessagePublishingErrorHandler();
+			}
+		}
 	}
 
 	@Override
@@ -154,7 +176,12 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 
 	@Override
 	public void onNext(Message<?> message) {
-		handleMessage(message);
+		try {
+			handleMessage(message);
+		}
+		catch (MessagingException e) {
+			this.reactiveErrorHandler.handleError(e);
+		}
 	}
 
 	@Override