GH-3986: Integration mock: fix ReactiveMH support

Fixes https://github.com/spring-projects/spring-integration/issues/3986

The `ReactiveStreamsConsumer` in addition to plain `Subscriber` and `MessageHandler`
also supports a `ReactiveMessageHandler`, but `MockIntegrationContext` doesn't
handle a scenario when `ReactiveMessageHandler` is provided for consumer

* Rework the logic in the `MockIntegrationContext` to substitute a `ReactiveMessageHandler`
in the `ReactiveStreamsConsumer` which has a precedence in its logic over plain `Subscriber`.
* Wrap a plain `MessageHandler` mock into a `ReactiveMessageHandler` before substitution
in the `ReactiveStreamsConsumer`
* Reset `ReactiveStreamsConsumer.reactiveMessageHandler` with source `ReactiveMessageHandler`
or `null` respectively to an original `ReactiveStreamsConsumer` configuration
* Mention `ReactiveMessageHandler` use-case in the `testing.adoc`

**Cherry-pick to `5.5.x`**

spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 the original author or authors.
+ * Copyright 2017-2023 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.
@@ -23,6 +23,7 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.reactivestreams.Subscriber;
+import reactor.core.publisher.Mono;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
@@ -30,6 +31,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.EndpointId;
 import org.springframework.integration.annotation.Poller;
+import org.springframework.integration.annotation.Reactive;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
@@ -48,6 +50,7 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
@@ -271,6 +274,24 @@ public class MockMessageHandlerTests {
 	}
 
 
+	@Autowired
+	private MessageChannel reactiveInputChannel;
+
+	@Test
+	void reactiveMessageHandlerSubstitution() {
+		MockMessageHandler mockMessageHandler =
+				mockMessageHandler()
+						.handleNext(message -> {
+						});
+
+		this.mockIntegrationContext.substituteMessageHandlerFor("reactiveEndpoint", mockMessageHandler);
+
+		this.reactiveInputChannel.send(new GenericMessage<>("test"));
+
+		verify(mockMessageHandler).handleMessage(any(Message.class));
+	}
+
+
 	@Configuration
 	@EnableIntegration
 	public static class Config {
@@ -341,6 +362,14 @@ public class MockMessageHandlerTests {
 			return new LoggingHandler(LoggingHandler.Level.FATAL);
 		}
 
+
+		@Bean
+		@EndpointId("reactiveEndpoint")
+		@ServiceActivator(inputChannel = "reactiveInputChannel", reactive = @Reactive)
+		public ReactiveMessageHandler reactiveMessageHandler() {
+			return message -> Mono.empty();
+		}
+
 	}
 
 }