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

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.AbstractMessageProducingHandler;
 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,16 +157,18 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 
 		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) {
 			return endpoint;
@@ -379,7 +382,13 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 			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.core.annotation.AnnotatedElementUtils;
 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 class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot
 	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 java.util.stream.Stream;
 
 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.ServiceActivator;
 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,6 +73,7 @@ import org.springframework.messaging.Message;
 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;
@@ -80,7 +81,11 @@ 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 @@ import org.springframework.test.context.junit4.SpringRunner;
  * @since 4.0
  */
 @ContextConfiguration(classes = MessagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.class)
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class MessagingAnnotationsWithBeanAnnotationTests {
 
@@ -225,6 +230,25 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
 				.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 class MessagingAnnotationsWithBeanAnnotationTests {
 			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