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

Fix ReactiveStreamsConsumer for plain subscriber

Browse Source 
https://build.spring.io/browse/INT-MASTERSPRING40-978

Investigate a behaviour for `ReactiveStreamsConsumer`
when we use a plain `Subscriber` directly instead of `doOn...` callbacks.
It looks like there is some race condition when the data can be consumed
upstream, but there is no consumer downstream ready yet
This commit is contained in:
Artem Bilan
2020-02-10 16:00:14 -05:00
parent 538cfe4868
commit 6591ce90f6
1 changed files with 13 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -162,22 +162,23 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
this.lifecycleDelegate.start();
}
Flux<?> flux = null;
if (this.reactiveMessageHandler != null) {
flux = Flux.from(this.publisher)
.flatMap(this.reactiveMessageHandler::handleMessage);
}
else if (this.subscriber != null) {
flux = Flux.from(this.publisher)
.doOnSubscribe(this.subscriber::onSubscribe)
.doOnComplete(this.subscriber::onComplete)
.doOnNext(this.subscriber::onNext);
}
if (flux != null) {
this.subscription =
flux.onErrorContinue((ex, data) -> this.errorHandler.handleError(ex))
Flux.from(this.publisher)
.flatMap(this.reactiveMessageHandler::handleMessage)
.onErrorContinue(this::onError)
.subscribe();
}
else if (this.subscriber != null) {
Flux.from(this.publisher)
.doOnSubscribe((subs) -> this.subscription = subs::cancel)
.onErrorContinue(this::onError)
.subscribe(this.subscriber);
}
}
private void onError(Throwable ex, Object data) {
this.errorHandler.handleError(ex);
}
@Override