INT-4537: Fix RSConsumer for MockIntegrationCtx

JIRA: https://jira.spring.io/browse/INT-4537

Fixes #2582

* Rename `ReactiveStreamsConsumer.messageHandler` property to the
`handler` for consistency with other `IntegrationConsumer` s
* Do not wrap `Subscriber` into the `MessageHandler` if that one is
already a `MessageHandler`
* Fix `MockIntegrationContext` for the logic around `ReactiveStreamsConsumer`
where it is not enough just replace a `handler`, but we also need to do
that with the `subscriber`.
Luckily the `MockMessageHandler` is also a Reactive `Subscriber`
* Clean up `MockIntegrationContext.beans` in the end of `resetBeans()`
* Improve `testing.adoc`

**Cherry-pick to 5.0.x**

Author: artembilan

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

@@ -49,7 +49,7 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 
 	private final MessageChannel inputChannel;
 
-	private final MessageHandler messageHandler;
+	private final MessageHandler handler;
 
 	private final Publisher<Message<Object>> publisher;
 
@@ -83,10 +83,13 @@ public ReactiveStreamsConsumer(MessageChannel inputChannel, final Subscriber<Mes
 		this.subscriber = subscriber;
 		this.lifecycleDelegate = subscriber instanceof Lifecycle ? (Lifecycle) subscriber : null;
 		if (subscriber instanceof MessageHandlerSubscriber) {
-			this.messageHandler = ((MessageHandlerSubscriber) subscriber).messageHandler;
+			this.handler = ((MessageHandlerSubscriber) subscriber).messageHandler;
+		}
+		else if (subscriber instanceof MessageHandler) {
+			this.handler = (MessageHandler) subscriber;
 		}
 		else {
-			this.messageHandler = this.subscriber::onNext;
+			this.handler = this.subscriber::onNext;
 		}
 	}
 
@@ -101,11 +104,11 @@ public MessageChannel getInputChannel() {
 
 	@Override
 	public MessageChannel getOutputChannel() {
-		if (this.messageHandler instanceof MessageProducer) {
-			return ((MessageProducer) this.messageHandler).getOutputChannel();
+		if (this.handler instanceof MessageProducer) {
+			return ((MessageProducer) this.handler).getOutputChannel();
 		}
-		else if (this.messageHandler instanceof MessageRouter) {
-			return ((MessageRouter) this.messageHandler).getDefaultOutputChannel();
+		else if (this.handler instanceof MessageRouter) {
+			return ((MessageRouter) this.handler).getDefaultOutputChannel();
 		}
 		else {
 			return null;
@@ -114,7 +117,7 @@ else if (this.messageHandler instanceof MessageRouter) {
 
 	@Override
 	public MessageHandler getHandler() {
-		return this.messageHandler;
+		return this.handler;
 	}
 
 	@Override

spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java

@@ -30,6 +30,7 @@
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.endpoint.IntegrationConsumer;
+import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.test.mock.MockMessageHandler;
 import org.springframework.integration.test.util.TestUtils;
@@ -38,6 +39,9 @@
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
+
 /**
  * A {@link BeanFactoryAware} component with an API to customize real beans
  * in the application context from test code.
@@ -91,10 +95,17 @@ public void resetBeans(String... beanNames) {
 					if (endpoint instanceof SourcePollingChannelAdapter) {
 						directFieldAccessor.setPropertyValue("source", e.getValue());
 					}
+					else if (endpoint instanceof ReactiveStreamsConsumer) {
+						Tuple2<?, ?> value = (Tuple2<?, ?>) e.getValue();
+						directFieldAccessor.setPropertyValue("handler", value.getT1());
+						directFieldAccessor.setPropertyValue("subscriber", value.getT2());
+					}
 					else if (endpoint instanceof IntegrationConsumer) {
 						directFieldAccessor.setPropertyValue("handler", e.getValue());
 					}
 				});
+
+		this.beans.clear();
 	}
 
 	/**
@@ -137,7 +148,13 @@ public void substituteMessageHandlerFor(String consumerEndpointId, MessageHandle
 		}
 		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(endpoint);
 		Object targetMessageHandler = directFieldAccessor.getPropertyValue("handler");
-		this.beans.put(consumerEndpointId, targetMessageHandler);
+		if (endpoint instanceof ReactiveStreamsConsumer) {
+			Object targetSubscriber = directFieldAccessor.getPropertyValue("subscriber");
+			this.beans.put(consumerEndpointId, Tuples.of(targetMessageHandler, targetSubscriber));
+		}
+		else {
+			this.beans.put(consumerEndpointId, targetMessageHandler);
+		}
 
 		if (mockMessageHandler instanceof MessageProducer) {
 			if (targetMessageHandler instanceof MessageProducer) {
@@ -160,6 +177,10 @@ public void substituteMessageHandlerFor(String consumerEndpointId, MessageHandle
 
 		directFieldAccessor.setPropertyValue("handler", mockMessageHandler);
 
+		if (endpoint instanceof ReactiveStreamsConsumer) {
+			directFieldAccessor.setPropertyValue("subscriber", mockMessageHandler);
+		}
+
 		if (autoStartup && endpoint instanceof Lifecycle) {
 			((Lifecycle) endpoint).start();
 		}

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

@@ -20,6 +20,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
@@ -36,6 +37,7 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
+import org.reactivestreams.Subscriber;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
@@ -46,7 +48,7 @@
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
-import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageHandler;
 import org.springframework.integration.support.MessageBuilder;
@@ -183,7 +185,8 @@ public void testMockRawHandler() {
 		ArgumentCaptor<Message<?>> messageArgumentCaptor = MockIntegration.messageArgumentCaptor();
 		MessageHandler mockMessageHandler =
 				spy(mockMessageHandler(messageArgumentCaptor))
-						.handleNext(m -> { });
+						.handleNext(m -> {
+						});
 
 		String endpointId = "rawHandlerConsumer";
 		this.mockIntegrationContext.substituteMessageHandlerFor(endpointId, mockMessageHandler);
@@ -214,6 +217,11 @@ public void testMockRawHandler() {
 			assertThat(e, instanceOf(IllegalStateException.class));
 			assertThat(e.getMessage(), containsString("with replies can't replace simple MessageHandler"));
 		}
+
+		this.mockIntegrationContext.resetBeans();
+
+		assertNotSame(mockMessageHandler, TestUtils.getPropertyValue(endpoint, "handler", MessageHandler.class));
+		assertNotSame(mockMessageHandler, TestUtils.getPropertyValue(endpoint, "subscriber", Subscriber.class));
 	}
 
 	/**
@@ -272,9 +280,9 @@ public SubscribableChannel rawChannel() {
 		}
 
 		@Bean
-		public EventDrivenConsumer rawHandlerConsumer() {
-			return new EventDrivenConsumer(rawChannel(),
-					new ExpressionEvaluatingMessageHandler(new ValueExpression<>("test")));
+		public ReactiveStreamsConsumer rawHandlerConsumer() {
+			return new ReactiveStreamsConsumer(rawChannel(),
+					(MessageHandler) new ExpressionEvaluatingMessageHandler(new ValueExpression<>("test")));
 		}
 
 		@ServiceActivator(inputChannel = "startChannel", outputChannel = "nextChannel")
@@ -291,7 +299,8 @@ public ArgumentCaptor<Message<?>> argumentCaptorForOutputTest() {
 		@ServiceActivator(inputChannel = "nextChannel")
 		public MessageHandler handleNextInput() {
 			return mockMessageHandler(argumentCaptorForOutputTest())
-					.handleNext(m -> { });
+					.handleNext(m -> {
+					});
 		}
 
 	}

src/reference/asciidoc/testing.adoc

@@ -265,14 +265,29 @@ public void testMockMessageSource() {
 ----
 ====
 
+NOTE: The `mySourceEndpoint` refers here to the bean name of the `SourcePollingChannelAdapter` for which we replace the real `MessageSource` with our mock.
+Similarly the `MockIntegrationContext.substituteMessageHandlerFor()` expects a bean name for the `IntegrationConsumer`, which wraps a `MessageHandler` as an endpoint.
+
+After test is performed you can restore the state of endpoint beans to the real configuration using `MockIntegrationContext.resetBeans()`:
+
+====
+[source,java]
+----
+@After
+public void tearDown() {
+    this.mockIntegrationContext.resetBeans();
+}
+----
+====
+
 See the https://docs.spring.io/spring-integration/api/org/springframework/integration/test/context/MockIntegrationContext.html[Javadoc] for more information.
 
 [[testing-mocks]]
 === Integration Mocks
 
 The `org.springframework.integration.test.mock` package offers tools and utilities for mocking, stubbing, and verification of activity on Spring Integration components.
 The mocking functionality is fully based on and compatible with the well known Mockito Framework.
-(The current Mockito transitive dependency is on version 2.5.x.)
+(The current Mockito transitive dependency is on version 2.5.x or higher.)
 
 ==== MockIntegration