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 3644c058fa
commit 11fcda4728
4 changed files with 67 additions and 19 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

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