Infrastructure for ReactiveMessageHandler (#3137)

* Infrastructure for ReactiveMessageHandler

We have now a `ReactiveMongoDbStoringMessageHandler` which implements
a `ReactiveMessageHandler`, but not a `MessageHandler` for possible
deferred subscriptions to the returned Reactor type

We don't have a proper application context processing for this
new type of message handlers

* Change a  `ConsumerEndpointFactoryBean` to apply an `MH` and `RMH`
as possible types for handler
* Introduce a `ReactiveMessageHandlerAdapter` to wrap an `RMH`
into a `MH` for synchronous calls in the regular consumer endpoints
* Wrap an `RMH` into a `ReactiveMessageHandlerAdapter` for regular
endpoints and unwrap for `ReactiveStreamsConsumer`
* Add `RMH`-based ctor into `ReactiveStreamsConsumer` for target
reactive streams composition (`flatMap()` on the `RMH`)
* Remove a `DelegatingSubscriber` from the `ReactiveStreamsConsumer`
in favor of direct calls from the `doOnSubscribe()`, `doOnComplete()`
& `doOnNext()`
* Add an `onErrorContinue()` to handle per-message errors, but don't
cancel the whole source `Publisher`
* Use `Disposable` from the `subscribe()` to cancel in the `stop()`
- recommended way in Reactor
* Use `onErrorContinue()` in the `FluxMessageChannel` instead of
`try..catch` in the `doOnNext()` - for possible `onErrorStop()`
in the provided upstream `Publisher`
* Handle `RMH` in the `ServiceActivatorFactoryBean` as a direct handler
as well with wrapping into `ReactiveMessageHandlerAdapter` for return.
The `ConsumerEndpointFactoryBean` extracts an `RMH` from the adapter
for the `ReactiveStreamsConsumer` anyway
* Add XML parsing test for `ReactiveMongoDbStoringMessageHandler`
* Add `log4j-slf4j-impl` for all the test runtime since `slf4j-api`
comes as a transitive dependency from many places

* * Fix conflicts after rebasing to master

* * Fix typo in warn message
* Change `Assert.state()` to `Assert.isTrue()`
for `ConsumerEndpointFactoryBean.setHandler()`

* * Fix `ConsumerEndpointFactoryBean` when reactive and no advice-chain
* Fix race condition in the
`ReactiveMongoDbStoringMessageHandlerTests.testReactiveMongoMessageHandlerFromApplicationContext()`

* * Handle `ReactiveMessageHandler` in Java DSL.
Essentially request a wrapping into `ReactiveMessageHandlerAdapter`.
Describe such a requirements in the `ReactiveMessageHandlerAdapter` JavaDocs
* Some Java DSL test polishing
* Add Java DSL for `ReactiveMongoDbStoringMessageHandler`
* Propagate missed `ApplicationContext` population into an internally
created `ReactiveMongoTemplate` in the `ReactiveMongoDbStoringMessageHandler`

Author: artembilan

build.gradle

@@ -237,6 +237,7 @@ configure(javaProjects) { subproject ->
 
 		testRuntimeOnly 'org.apache.logging.log4j:log4j-core'
 		testRuntimeOnly 'org.apache.logging.log4j:log4j-jcl'
+		testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j-impl'
 	}
 
 	// enable all compiler warnings; individual projects may customize further
@@ -471,7 +472,6 @@ project('spring-integration-gemfire') {
 		api "commons-io:commons-io:$commonsIoVersion"
 
 		testImplementation project(':spring-integration-stream')
-		testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j-impl'
 	}
 }
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -78,14 +78,8 @@ public void subscribeTo(Publisher<? extends Message<?>> publisher) {
 				Flux.from(publisher)
 						.delaySubscription(this.subscribedSignal.filter(Boolean::booleanValue).next())
 						.publishOn(Schedulers.boundedElastic())
-						.doOnNext((message) -> {
-							try {
-								send(message);
-							}
-							catch (Exception e) {
-								logger.warn("Error during processing event: " + message, e);
-							}
-						})
+						.doOnNext(this::send)
+						.onErrorContinue((ex, message) -> logger.warn("Error during processing event: " + message, ex))
 						.subscribe());
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java

@@ -42,12 +42,14 @@
 import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
 import org.springframework.integration.handler.advice.HandleMessageAdvice;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.support.channel.ChannelResolverUtils;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.core.DestinationResolver;
 import org.springframework.scheduling.TaskScheduler;
@@ -112,11 +114,17 @@ public class ConsumerEndpointFactoryBean
 
 	private volatile boolean initialized;
 
-	public void setHandler(MessageHandler handler) {
-		Assert.notNull(handler, "handler must not be null");
+	public void setHandler(Object handler) {
+		Assert.isTrue(handler instanceof MessageHandler || handler instanceof ReactiveMessageHandler,
+				"'handler' must be an instance of 'MessageHandler' or 'ReactiveMessageHandler'");
 		synchronized (this.handlerMonitor) {
 			Assert.isNull(this.handler, "handler cannot be overridden");
-			this.handler = handler;
+			if (handler instanceof ReactiveMessageHandler) {
+				this.handler = new ReactiveMessageHandlerAdapter((ReactiveMessageHandler) handler);
+			}
+			else {
+				this.handler = (MessageHandler) handler;
+			}
 		}
 	}
 
@@ -210,7 +218,12 @@ public void afterPropertiesSet() {
 			}
 		}
 
-		adviceChain();
+		if (!(this.handler instanceof ReactiveMessageHandlerAdapter)) {
+			adviceChain();
+		}
+		else if (!CollectionUtils.isEmpty(this.adviceChain)) {
+			LOGGER.warn("the advice chain cannot be applied to a 'ReactiveMessageHandler'");
+		}
 		if (this.channelResolver == null) {
 			this.channelResolver = ChannelResolverUtils.getChannelResolver(this.beanFactory);
 		}
@@ -282,7 +295,13 @@ else if (channel instanceof PollableChannel) {
 				pollingConsumer(channel);
 			}
 			else {
-				this.endpoint = new ReactiveStreamsConsumer(channel, this.handler);
+				if (this.handler instanceof ReactiveMessageHandlerAdapter) {
+					this.endpoint = new ReactiveStreamsConsumer(channel,
+							((ReactiveMessageHandlerAdapter) this.handler).getDelegate());
+				}
+				else {
+					this.endpoint = new ReactiveStreamsConsumer(channel, this.handler);
+				}
 			}
 			this.endpoint.setBeanName(this.beanName);
 			this.endpoint.setBeanFactory(this.beanFactory);

spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,9 +22,11 @@
 import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.integration.handler.MessageProcessor;
+import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
 import org.springframework.integration.handler.ReplyProducingMessageHandlerWrapper;
 import org.springframework.integration.handler.ServiceActivatingHandler;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.util.StringUtils;
 
 /**
@@ -47,7 +49,7 @@ public void setNotPropagatedHeaders(String... headers) {
 
 	@Override
 	protected MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) {
-		MessageHandler handler = null;
+		MessageHandler handler;
 		handler = createDirectHandlerIfPossible(targetObject, targetMethodName);
 		if (handler == null) {
 			handler = configureHandler(
@@ -67,34 +69,38 @@ protected MessageHandler createMethodInvokingHandler(Object targetObject, String
 	 */
 	protected MessageHandler createDirectHandlerIfPossible(final Object targetObject, String targetMethodName) {
 		MessageHandler handler = null;
-		if (targetObject instanceof MessageHandler
-				&& this.methodIsHandleMessageOrEmpty(targetMethodName)) {
+		if ((targetObject instanceof MessageHandler || targetObject instanceof ReactiveMessageHandler)
+				&& methodIsHandleMessageOrEmpty(targetMethodName)) {
 			if (targetObject instanceof AbstractMessageProducingHandler) {
 				// should never happen but just return it if it's already an AMPH
 				return (MessageHandler) targetObject;
 			}
-			/*
-			 * Return a reply-producing message handler so that we still get 'produced no reply' messages
-			 * and the super class will inject the advice chain to advise the handler method if needed.
-			 */
-			handler = new ReplyProducingMessageHandlerWrapper((MessageHandler) targetObject);
-
+			if (targetObject instanceof ReactiveMessageHandler) {
+				handler = new ReactiveMessageHandlerAdapter((ReactiveMessageHandler) targetObject);
+			}
+			else {
+				/*
+				 * Return a reply-producing message handler so that we still get 'produced no reply' messages
+				 * and the super class will inject the advice chain to advise the handler method if needed.
+				 */
+				handler = new ReplyProducingMessageHandlerWrapper((MessageHandler) targetObject);
+			}
 		}
 		return handler;
 	}
 
 	@Override
 	protected MessageHandler createExpressionEvaluatingHandler(Expression expression) {
-		ExpressionEvaluatingMessageProcessor<Object> processor = new ExpressionEvaluatingMessageProcessor<Object>(expression);
-		processor.setBeanFactory(this.getBeanFactory());
+		ExpressionEvaluatingMessageProcessor<Object> processor = new ExpressionEvaluatingMessageProcessor<>(expression);
+		processor.setBeanFactory(getBeanFactory());
 		ServiceActivatingHandler handler = new ServiceActivatingHandler(processor);
 		handler.setPrimaryExpression(expression);
 		return this.configureHandler(handler);
 	}
 
 	@Override
 	protected <T> MessageHandler createMessageProcessingHandler(MessageProcessor<T> processor) {
-		return this.configureHandler(new ServiceActivatingHandler(processor));
+		return configureHandler(new ServiceActivatingHandler(processor));
 	}
 
 	protected MessageHandler configureHandler(ServiceActivatingHandler handler) {
@@ -115,7 +121,6 @@ protected boolean canBeUsedDirect(AbstractMessageProducingHandler handler) {
 	@Override
 	protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) {
 		super.postProcessReplyProducer(handler);
-
 		if (this.headers != null) {
 			handler.setNotPropagatedHeaders(this.headers);
 		}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,19 +27,25 @@
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
 import org.springframework.integration.router.MessageRouter;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
 
 import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
-import reactor.core.publisher.BaseSubscriber;
+import reactor.core.publisher.Flux;
 
 
 /**
+ * An {@link AbstractEndpoint} implementation for Reactive Streams subscription into an
+ * input channel and reactive consumption of messages from that channel.
+ *
  * @author Artem Bilan
  *
  * @since 5.0
@@ -48,17 +54,22 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 
 	private final MessageChannel inputChannel;
 
+	private final Publisher<Message<Object>> publisher;
+
 	private final MessageHandler handler;
 
-	private final Publisher<Message<Object>> publisher;
+	@Nullable
+	private final ReactiveMessageHandler reactiveMessageHandler;
 
+	@Nullable
 	private final Subscriber<Message<?>> subscriber;
 
+	@Nullable
 	private final Lifecycle lifecycleDelegate;
 
 	private ErrorHandler errorHandler;
 
-	private volatile Subscription subscription;
+	private volatile Disposable subscription;
 
 	@SuppressWarnings("unchecked")
 	public ReactiveStreamsConsumer(MessageChannel inputChannel, MessageHandler messageHandler) {
@@ -68,10 +79,10 @@ public ReactiveStreamsConsumer(MessageChannel inputChannel, MessageHandler messa
 						: new MessageHandlerSubscriber(messageHandler));
 	}
 
-	public ReactiveStreamsConsumer(MessageChannel inputChannel, final Subscriber<Message<?>> subscriber) {
-		this.inputChannel = inputChannel;
+	public ReactiveStreamsConsumer(MessageChannel inputChannel, Subscriber<Message<?>> subscriber) {
 		Assert.notNull(inputChannel, "'inputChannel' must not be null");
 		Assert.notNull(subscriber, "'subscriber' must not be null");
+		this.inputChannel = inputChannel;
 
 		if (inputChannel instanceof NullChannel && logger.isWarnEnabled()) {
 			logger.warn("The consuming from the NullChannel does not have any effects: " +
@@ -90,6 +101,24 @@ else if (subscriber instanceof MessageHandler) {
 		else {
 			this.handler = this.subscriber::onNext;
 		}
+		this.reactiveMessageHandler = null;
+	}
+
+	/**
+	 * Instantiate an endpoint based on the provided {@link MessageChannel} and {@link ReactiveMessageHandler}.
+	 * @param inputChannel the channel to consume in reactive manner.
+	 * @param reactiveMessageHandler the {@link ReactiveMessageHandler} to process messages.
+	 * @since 5.3
+	 */
+	public ReactiveStreamsConsumer(MessageChannel inputChannel, ReactiveMessageHandler reactiveMessageHandler) {
+		Assert.notNull(inputChannel, "'inputChannel' must not be null");
+		this.inputChannel = inputChannel;
+		this.handler = new ReactiveMessageHandlerAdapter(reactiveMessageHandler);
+		this.reactiveMessageHandler = reactiveMessageHandler;
+		this.publisher = MessageChannelReactiveUtils.toPublisher(inputChannel);
+		this.subscriber = null;
+		this.lifecycleDelegate =
+				reactiveMessageHandler instanceof Lifecycle ? (Lifecycle) reactiveMessageHandler : null;
 	}
 
 	public void setErrorHandler(ErrorHandler errorHandler) {
@@ -132,55 +161,35 @@ protected void doStart() {
 		if (this.lifecycleDelegate != null) {
 			this.lifecycleDelegate.start();
 		}
-		this.publisher.subscribe(new DelegatingSubscriber());
+
+		Flux<?> flux = null;
+		if (this.reactiveMessageHandler != null) {
+			flux = Flux.from(this.publisher)
+					.flatMap(this.reactiveMessageHandler::handleMessage);
+		}
+		else if (this.subscriber != null) {
+			flux = Flux.from(this.publisher)
+					.doOnSubscribe(this.subscriber::onSubscribe)
+					.doOnComplete(this.subscriber::onComplete)
+					.doOnNext(this.subscriber::onNext);
+		}
+		if (flux != null) {
+			this.subscription =
+					flux.onErrorContinue((ex, data) -> this.errorHandler.handleError(ex))
+							.subscribe();
+		}
 	}
 
 	@Override
 	protected void doStop() {
 		if (this.subscription != null) {
-			this.subscription.cancel();
+			this.subscription.dispose();
 		}
 		if (this.lifecycleDelegate != null) {
 			this.lifecycleDelegate.stop();
 		}
 	}
 
-	private final class DelegatingSubscriber extends BaseSubscriber<Message<?>> {
-
-		private final Subscriber<Message<?>> delegate = ReactiveStreamsConsumer.this.subscriber;
-
-		DelegatingSubscriber() {
-		}
-
-		@Override
-		public void hookOnSubscribe(Subscription s) {
-			ReactiveStreamsConsumer.this.subscription = s;
-			this.delegate.onSubscribe(s);
-		}
-
-		@Override
-		public void hookOnNext(Message<?> message) {
-			try {
-				this.delegate.onNext(message);
-			}
-			catch (Exception e) {
-				ReactiveStreamsConsumer.this.errorHandler.handleError(e);
-				hookOnError(e);
-			}
-		}
-
-		@Override
-		public void hookOnError(Throwable t) {
-			this.delegate.onError(t);
-		}
-
-		@Override
-		public void hookOnComplete() {
-			this.delegate.onComplete();
-		}
-
-	}
-
 	private static final class MessageHandlerSubscriber
 			implements CoreSubscriber<Message<?>>, Disposable, Lifecycle {