From 2df71fba3046aa0e259eb14a53fbd22cccc1cafa Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 5 Oct 2018 13:36:29 -0400 Subject: [PATCH] INT-4537: Fix RSConsumer for MockIntegrationCtx JIRA: https://jira.spring.io/browse/INT-4537 Fixes spring-projects/spring-integration#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** --- .../endpoint/ReactiveStreamsConsumer.java | 19 ++++++++------- .../test/context/MockIntegrationContext.java | 23 ++++++++++++++++++- .../test/mock/MockMessageHandlerTests.java | 21 ++++++++++++----- src/reference/asciidoc/testing.adoc | 17 +++++++++++++- 4 files changed, 64 insertions(+), 16 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java index cd8b90e6ea..aeb2c8545f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java +++ b/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> publisher; @@ -83,10 +83,13 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra 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 class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra @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 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra @Override public MessageHandler getHandler() { - return this.messageHandler; + return this.handler; } @Override diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java b/spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java index 3c9745d64d..14782eb9a3 100644 --- a/spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java @@ -30,6 +30,7 @@ import org.springframework.context.Lifecycle; 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.messaging.MessageHandler; 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 class MockIntegrationContext implements BeanFactoryAware { 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 class MockIntegrationContext implements BeanFactoryAware { } 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 class MockIntegrationContext implements BeanFactoryAware { directFieldAccessor.setPropertyValue("handler", mockMessageHandler); + if (endpoint instanceof ReactiveStreamsConsumer) { + directFieldAccessor.setPropertyValue("subscriber", mockMessageHandler); + } + if (autoStartup && endpoint instanceof Lifecycle) { ((Lifecycle) endpoint).start(); } diff --git a/spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageHandlerTests.java b/spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageHandlerTests.java index 7aebcf4d09..b98a6460e7 100644 --- a/spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageHandlerTests.java +++ b/spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageHandlerTests.java @@ -20,6 +20,7 @@ import static org.hamcrest.Matchers.containsString; 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.After; 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.annotation.ServiceActivator; 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 class MockMessageHandlerTests { ArgumentCaptor> 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 class MockMessageHandlerTests { 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 class MockMessageHandlerTests { } @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 class MockMessageHandlerTests { @ServiceActivator(inputChannel = "nextChannel") public MessageHandler handleNextInput() { return mockMessageHandler(argumentCaptorForOutputTest()) - .handleNext(m -> { }); + .handleNext(m -> { + }); } } diff --git a/src/reference/asciidoc/testing.adoc b/src/reference/asciidoc/testing.adoc index 4c4e548838..9cabb40328 100644 --- a/src/reference/asciidoc/testing.adoc +++ b/src/reference/asciidoc/testing.adoc @@ -265,6 +265,21 @@ 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]] @@ -272,7 +287,7 @@ See the https://docs.spring.io/spring-integration/api/org/springframework/integr 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