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

@@ -47,7 +47,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;
 
@@ -76,10 +76,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;
 		}
 	}
 
@@ -94,11 +97,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;
@@ -107,7 +110,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

@@ -229,14 +229,29 @@ public void testMockMessageSource() {
 }
 ----
 
-See their JavaDocs for more information.
+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 and compatible with the well known Mockito Framework.
-(The current Mockito transitive dependency is of _version 2.5.x_.)
+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 or higher.)
 
 ==== MockIntegration