Messaging Annotations: process ReactiveMH (#3141)

* Messaging Annotations: process ReactiveMH

* Add support for `ReactiveMessageHandler` as a `@ServiceActivator` bean
* Wrap `ReactiveMessageHandler` in the `ReactiveMessageHandlerAdapter`
in the `ServiceActivatorAnnotationPostProcessor`
* Unwrap `ReactiveMessageHandlerAdapter` in the `AbstractMethodAnnotationPostProcessor`
for `ReactiveStreamsConsumer`

* * Add documentation for the `ReactiveMessageHandler`
& Reactive MongoDb channel adapters

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.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.
@@ -66,6 +66,7 @@
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.LambdaMessageProcessor;
 import org.springframework.integration.handler.MessageProcessor;
+import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
 import org.springframework.integration.handler.ReplyProducingMessageHandlerWrapper;
 import org.springframework.integration.handler.advice.HandleMessageAdvice;
 import org.springframework.integration.router.AbstractMessageRouter;
@@ -104,7 +105,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 
 	protected static final String SEND_TIMEOUT_ATTRIBUTE = "sendTimeout";
 
-	protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
+	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR
 
 	protected final List<String> messageHandlerAttributes = new ArrayList<>(); // NOSONAR
 
@@ -156,15 +157,17 @@ public Object postProcess(Object bean, String beanName, Method method, List<Anno
 
 		MessageHandler handler = createHandler(bean, method, annotations);
 
-		orderable(method, handler);
-		producerOrRouter(annotations, handler);
+		if (!(handler instanceof ReactiveMessageHandlerAdapter)) {
+			orderable(method, handler);
+			producerOrRouter(annotations, handler);
 
-		if (!handler.equals(sourceHandler)) {
-			handler = registerHandlerBean(beanName, method, handler);
-		}
+			if (!handler.equals(sourceHandler)) {
+				handler = registerHandlerBean(beanName, method, handler);
+			}
 
-		handler = annotated(method, handler);
-		handler = adviceChain(beanName, annotations, handler);
+			handler = annotated(method, handler);
+			handler = adviceChain(beanName, annotations, handler);
+		}
 
 		AbstractEndpoint endpoint = createEndpoint(handler, method, annotations);
 		if (endpoint != null) {
@@ -379,7 +382,13 @@ protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChann
 			Assert.state(ObjectUtils.isEmpty(pollers), "A '@Poller' should not be specified for Annotation-based " +
 					"endpoint, since '" + inputChannel + "' is a SubscribableChannel (not pollable).");
 			if (inputChannel instanceof Publisher) {
-				endpoint = new ReactiveStreamsConsumer(inputChannel, handler);
+				if (handler instanceof ReactiveMessageHandlerAdapter) {
+					endpoint = new ReactiveStreamsConsumer(inputChannel,
+							((ReactiveMessageHandlerAdapter) handler).getDelegate());
+				}
+				else {
+					endpoint = new ReactiveStreamsConsumer(inputChannel, handler);
+				}
 			}
 			else {
 				endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.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.
@@ -27,10 +27,12 @@
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 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.integration.util.MessagingAnnotationUtils;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.util.StringUtils;
 
 /**
@@ -53,9 +55,12 @@ public ServiceActivatorAnnotationPostProcessor(ConfigurableListableBeanFactory b
 	protected MessageHandler createHandler(Object bean, Method method, List<Annotation> annotations) {
 		AbstractReplyProducingMessageHandler serviceActivator;
 		if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
-			final Object target = resolveTargetBeanFromMethodWithBeanAnnotation(method);
+			Object target = resolveTargetBeanFromMethodWithBeanAnnotation(method);
 			serviceActivator = extractTypeIfPossible(target, AbstractReplyProducingMessageHandler.class);
 			if (serviceActivator == null) {
+				if (target instanceof ReactiveMessageHandler) {
+					return new ReactiveMessageHandlerAdapter((ReactiveMessageHandler) target);
+				}
 				if (target instanceof MessageHandler) {
 					/*
 					 * Return a reply-producing message handler so that we still get 'produced no reply' messages

spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-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.
@@ -29,8 +29,7 @@
 
 import javax.annotation.Resource;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -56,6 +55,7 @@
 import org.springframework.integration.annotation.Splitter;
 import org.springframework.integration.annotation.Transformer;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.EnableMessageHistory;
@@ -73,14 +73,19 @@
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+import reactor.test.StepVerifier;
 
 /**
  * @author Artem Bilan
@@ -90,7 +95,7 @@
  * @since 4.0
  */
 @ContextConfiguration(classes = MessagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.class)
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class MessagingAnnotationsWithBeanAnnotationTests {
 
@@ -225,6 +230,25 @@ public void testInvalidMessagingAnnotationsConfig() {
 				.withMessageContaining("The attribute causing the ambiguity is: [applySequence].");
 	}
 
+	@Autowired
+	private MessageChannel reactiveMessageHandlerChannel;
+
+	@Autowired
+	private ContextConfiguration contextConfiguration;
+
+	@Test
+	public void testReactiveMessageHandler() {
+		this.reactiveMessageHandlerChannel.send(new GenericMessage<>("test"));
+
+		StepVerifier.create(
+				this.contextConfiguration.messageMonoProcessor
+						.map(Message::getPayload)
+						.cast(String.class))
+				.expectNext("test")
+				.verifyComplete();
+	}
+
+
 	@Configuration
 	@EnableIntegration
 	@EnableMessageHistory
@@ -403,6 +427,23 @@ public Consumer<Message<?>> messageConsumerAsService() {
 			return collector()::add;
 		}
 
+		MonoProcessor<Message<?>> messageMonoProcessor = MonoProcessor.create();
+
+		@Bean
+		MessageChannel reactiveMessageHandlerChannel() {
+			return new FluxMessageChannel();
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "reactiveMessageHandlerChannel")
+		public ReactiveMessageHandler reactiveMessageHandlerService() {
+			return (message) -> {
+				messageMonoProcessor.onNext(message);
+				messageMonoProcessor.onComplete();
+				return Mono.empty();
+			};
+		}
+
 	}
 
 	@Configuration

src/reference/asciidoc/mongodb.adoc

@@ -467,3 +467,29 @@ private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway() {
     }
 ----
 ====
+
+[[mongodb-reactive-channel-adapters]]
+=== MongoDB Reactive Channel Adapters
+
+Starting with version 5.3, the `ReactiveMongoDbStoringMessageHandler` implementation is provided.
+It is based on the `ReactiveMongoOperations` from Spring Data and requires a `org.mongodb:mongodb-driver-reactivestreams` dependency.
+This is an implementation of the `ReactiveMessageHandler` which is supported natively in the framework when reactive streams composition is involved in the integration flow definition.
+See more information in the <<./reactive-streams.adoc/reactive-message-handler,ReactiveMessageHandler>>.
+
+From configuration perspective there is no difference with many other standard channel adapters.
+For example with Java DSL such a channel adapter could be used like:
+
+====
+[source, java]
+----
+@Bean
+public IntegrationFlow reactiveMongoDbFlow(ReactiveMongoDatabaseFactory mongoDbFactory) {
+    return f -> f
+            .channel(MessageChannels.flux())
+            .handle(MongoDb.reactiveOutboundChannelAdapter(mongoDbFactory));
+}
+----
+====
+
+In this sample we are going to connect to the MongoDb via provided `ReactiveMongoDatabaseFactory` and store a data from request message into a default collection with the `data` name.
+The real operation is going to be performed on-demand from the reactive stream composition in the internally created `ReactiveStreamsConsumer`.

src/reference/asciidoc/reactive-streams.adoc

@@ -105,6 +105,18 @@ A result of the function is wrapped into a `Mono<Message<?>>` for flat-mapping i
 
 See <<./dsl.adoc#java-dsl,Java DSL Chapter>> for more information.
 
+[[reactive-message-handler]]
+=== `ReactiveMessageHandler`
+
+Starting with version 5.3, the `ReactiveMessageHandler` is supported natively in the framework.
+This type of message handler is designed for reactive clients which return a reactive type for on-demand subscription for low-level operation execution and doesn't provide any reply data to continue a reactive stream composition.
+When a `ReactiveMessageHandler` is used in the imperative integration flow, the `handleMessage()` result in subscribed immediately after return, just because there is no reactive streams composition in such a flow to honor back-pressure.
+In this case the framework wraps this `ReactiveMessageHandler` into a `ReactiveMessageHandlerAdapter` - a plain implementation of `MessageHandler`.
+However when a `ReactiveStreamsConsumer` is involved in the flow (e.g. when channel to consume is a `FluxMessageChannel`), such a `ReactiveMessageHandler` is composed to the whole reactive stream with a `flatMap()` Reactor operator to honor back-pressure during consumption.
+
+One of the out-of-the-box `ReactiveMessageHandler` implementation is a `ReactiveMongoDbStoringMessageHandler` for Outbound Channel Adapter.
+See <<./mongodb.adoc#mongodb-reactive-channel-adapters,MongoDB Reactive Channel Adapters>> for more information.
+
 [[reactive-channel-adapters]]
 === Reactive Channel Adapters
 
@@ -120,7 +132,7 @@ A reactive outbound channel adapter implementation is about initiation (or conti
 An inbound payload could be a reactive type per se or as an event of the whole integration flow which is a part of reactive stream on top.
 A returned reactive type can be subscribed immediately if we are in one-way, fire-and-forget scenario, or it is propagated downstream (request-reply scenarios) for further integration flow or an explicit subscription in the target business logic, but still downstream preserving reactive streams semantics.
 
-Currently Spring Integration provides channel adapter (or gateway) implementations for <<./webflux.adoc#webflux,WebFlux>> and <<./rsocket.adoc#rsocket,RSocket>>.
+Currently Spring Integration provides channel adapter (or gateway) implementations for <<./webflux.adoc#webflux,WebFlux>>, <<./rsocket.adoc#rsocket,RSocket>> and <<./mongodb.adoc#mongodb,MongoDb>>.
 Also an https://github.com/spring-projects/spring-integration-extensions/tree/master/spring-integration-cassandra[Apache Cassandra Extension] provides a `MessageHandler` implementation for the Cassandra reactive driver.
 More reactive channel adapters are coming, for example for https://r2dbc.io/[R2DBC], https://mongodb.github.io/mongo-java-driver-reactivestreams/[MongoDB], for Apache Kafka in https://github.com/spring-projects/spring-integration-kafka[Spring Integration Kafka] based on the `ReactiveKafkaProducerTemplate` and `ReactiveKafkaConsumerTemplate` from https://spring.io/projects/spring-kafka[Spring for Apache Kafka] etc.
 For many other non-reactive channel adapters thread pools are recommended to avoid blocking during reactive stream processing.

src/reference/asciidoc/whats-new.adoc

@@ -21,11 +21,23 @@ If you are interested in more details, see the Issue Tracker tickets that were r
 The `IntegrationPattern` abstraction has been introduced to indicate which enterprise integration pattern (an `IntegrationPatternType`) and category a Spring Integration component belongs to.
 See its JavaDocs and <<./graph.adoc#integration-graph,Integration Graph>> for more information about this abstraction and its use-cases.
 
+[[x5.3-reactive-message-handler]]
+==== `ReactiveMessageHandler`
+
+The `ReactiveMessageHandler` is now natively supported in the framework.
+See <<./reactive-streams.adoc/reactive-message-handler,ReactiveMessageHandler>> for more information.
+
+[[x5.3-mongodb-reactive-channel-adapters]]
+==== MongoDB Reactive Channel Adapters
+
+`spring-integration-mongodb` module now provides channel adapter implementations for Reactive MongoDB driver support in Spring Data.
+See <<./mongodb.adoc#mongodb-reactive-channel-adapters,MongoDB Reactive Channel Adapters>> for more information.
+
 [[x5.3-general]]
 === General Changes
 
 The gateway proxy now doesn't proxy `default` methods by default.
-see <<./gateway.adoc/gateway-calling-default-methods,Invoking `default` Methods>> for more information.
+See <<./gateway.adoc/gateway-calling-default-methods,Invoking `default` Methods>> for more information.
 
 
 Internal components (such as `_org.springframework.integration.errorLogger`) now have a shortened name when they are represented in the integration graph.