Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4537: Fix RSConsumer for MockIntegrationCtx

Browse Source 
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**
This commit is contained in:
Artem Bilan
2018-10-05 13:36:29 -04:00
committed by Gary Russell
parent 9e9fa2cf58
commit 2df71fba30
4 changed files with 64 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 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

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

@@ -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();
}

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

@@ -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<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 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 -> {
});
}
}

17
src/reference/asciidoc/testing.adoc
View File

@@ -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