GH-3986: Integration mock: fix ReactiveMH support
Fixes https://github.com/spring-projects/spring-integration/issues/3986 The `ReactiveStreamsConsumer` in addition to plain `Subscriber` and `MessageHandler` also supports a `ReactiveMessageHandler`, but `MockIntegrationContext` doesn't handle a scenario when `ReactiveMessageHandler` is provided for consumer * Rework the logic in the `MockIntegrationContext` to substitute a `ReactiveMessageHandler` in the `ReactiveStreamsConsumer` which has a precedence in its logic over plain `Subscriber`. * Wrap a plain `MessageHandler` mock into a `ReactiveMessageHandler` before substitution in the `ReactiveStreamsConsumer` * Reset `ReactiveStreamsConsumer.reactiveMessageHandler` with source `ReactiveMessageHandler` or `null` respectively to an original `ReactiveStreamsConsumer` configuration * Mention `ReactiveMessageHandler` use-case in the `testing.adoc` **Cherry-pick to `5.5.x`**
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2017-2021 the original author or authors.
|
||||
* Copyright 2017-2023 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -37,9 +37,11 @@ import org.springframework.integration.test.mock.MockMessageHandler;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.ReactiveMessageHandler;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.util.function.Tuple2;
|
||||
import reactor.util.function.Tuples;
|
||||
|
||||
@@ -118,9 +120,15 @@ public class MockIntegrationContext implements BeanFactoryAware {
|
||||
directFieldAccessor.setPropertyValue("source", handler);
|
||||
}
|
||||
else if (endpoint instanceof ReactiveStreamsConsumer) {
|
||||
Tuple2<?, ?> value = (Tuple2<?, ?>) handler;
|
||||
directFieldAccessor.setPropertyValue(HANDLER, value.getT1());
|
||||
directFieldAccessor.setPropertyValue("subscriber", value.getT2());
|
||||
if (handler instanceof Tuple2<?, ?>) {
|
||||
Tuple2<?, ?> value = (Tuple2<?, ?>) handler;
|
||||
directFieldAccessor.setPropertyValue(HANDLER, value.getT1());
|
||||
directFieldAccessor.setPropertyValue("reactiveMessageHandler", value.getT2());
|
||||
}
|
||||
else {
|
||||
directFieldAccessor.setPropertyValue(HANDLER, handler);
|
||||
directFieldAccessor.setPropertyValue("reactiveMessageHandler", null);
|
||||
}
|
||||
}
|
||||
else if (endpoint instanceof IntegrationConsumer) {
|
||||
directFieldAccessor.setPropertyValue(HANDLER, handler);
|
||||
@@ -173,9 +181,13 @@ public class MockIntegrationContext implements BeanFactoryAware {
|
||||
Object targetMessageHandler = directFieldAccessor.getPropertyValue(HANDLER);
|
||||
Assert.notNull(targetMessageHandler, () -> "'handler' must not be null in the: " + endpoint);
|
||||
if (endpoint instanceof ReactiveStreamsConsumer) {
|
||||
Object targetSubscriber = directFieldAccessor.getPropertyValue("subscriber");
|
||||
Assert.notNull(targetSubscriber, () -> "'subscriber' must not be null in the: " + endpoint);
|
||||
this.beans.put(consumerEndpointId, Tuples.of(targetMessageHandler, targetSubscriber));
|
||||
Object targetReactiveMessageHandler = directFieldAccessor.getPropertyValue("reactiveMessageHandler");
|
||||
if (targetReactiveMessageHandler != null) {
|
||||
this.beans.put(consumerEndpointId, Tuples.of(targetMessageHandler, targetReactiveMessageHandler));
|
||||
}
|
||||
else {
|
||||
this.beans.put(consumerEndpointId, targetMessageHandler);
|
||||
}
|
||||
}
|
||||
else {
|
||||
this.beans.put(consumerEndpointId, targetMessageHandler);
|
||||
@@ -203,7 +215,9 @@ public class MockIntegrationContext implements BeanFactoryAware {
|
||||
directFieldAccessor.setPropertyValue(HANDLER, mockMessageHandler);
|
||||
|
||||
if (endpoint instanceof ReactiveStreamsConsumer) {
|
||||
directFieldAccessor.setPropertyValue("subscriber", mockMessageHandler);
|
||||
ReactiveMessageHandler reactiveMessageHandler =
|
||||
(message) -> Mono.fromRunnable(() -> mockMessageHandler.handleMessage(message));
|
||||
directFieldAccessor.setPropertyValue("reactiveMessageHandler", reactiveMessageHandler);
|
||||
}
|
||||
|
||||
if (autoStartup && endpoint instanceof Lifecycle) {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2017-2020 the original author or authors.
|
||||
* Copyright 2017-2023 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -38,6 +38,7 @@ import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.annotation.EndpointId;
|
||||
import org.springframework.integration.annotation.Poller;
|
||||
import org.springframework.integration.annotation.Reactive;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
@@ -56,12 +57,15 @@ import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
import org.springframework.messaging.ReactiveMessageHandler;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
|
||||
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
* @author Yicheng Feng
|
||||
@@ -269,6 +273,24 @@ public class MockMessageHandlerTests {
|
||||
}
|
||||
|
||||
|
||||
@Autowired
|
||||
private MessageChannel reactiveInputChannel;
|
||||
|
||||
@Test
|
||||
void reactiveMessageHandlerSubstitution() {
|
||||
MockMessageHandler mockMessageHandler =
|
||||
mockMessageHandler()
|
||||
.handleNext(message -> {
|
||||
});
|
||||
|
||||
this.mockIntegrationContext.substituteMessageHandlerFor("reactiveEndpoint", mockMessageHandler);
|
||||
|
||||
this.reactiveInputChannel.send(new GenericMessage<>("test"));
|
||||
|
||||
verify(mockMessageHandler).handleMessage(any(Message.class));
|
||||
}
|
||||
|
||||
|
||||
@Configuration
|
||||
@EnableIntegration
|
||||
public static class Config {
|
||||
@@ -339,6 +361,14 @@ public class MockMessageHandlerTests {
|
||||
return new LoggingHandler(LoggingHandler.Level.FATAL);
|
||||
}
|
||||
|
||||
|
||||
@Bean
|
||||
@EndpointId("reactiveEndpoint")
|
||||
@ServiceActivator(inputChannel = "reactiveInputChannel", reactive = @Reactive)
|
||||
public ReactiveMessageHandler reactiveMessageHandler() {
|
||||
return message -> Mono.empty();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -347,6 +347,8 @@ assertSame(message, messageArgumentCaptor.getValue());
|
||||
----
|
||||
====
|
||||
|
||||
NOTE: The regular `MessageHandler` mocking (or `MockMessageHandler`) has to be used even for a `ReactiveStreamsConsumer` with a `ReactiveMessageHandler` configuration.
|
||||
|
||||
See the https://docs.spring.io/spring-integration/api/org/springframework/integration/test/mock/MockIntegration.html[`MockIntegration`] and https://docs.spring.io/spring-integration/api/org/springframework/integration/test/mock/MockMessageHandler.html[`MockMessageHandler`] Javadoc for more information.
|
||||
|
||||
[[testing-other-resources]]
|
||||
|
||||
Reference in New Issue
Block a user