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 4e29b0ce87..b34c131706 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 @@ -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> publisher; @@ -76,10 +76,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; } } @@ -94,11 +97,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; @@ -107,7 +110,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 4487408c21..1c0e21f742 100644 --- a/src/reference/asciidoc/testing.adoc +++ b/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