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

Fix ReactiveStreamsConsumer for error handling

Browse Source 
To support `onErrorContinue()` logic for the plain `Subscriber`
we need to wrap its `onNext()` into a `try..catch` and respective
`errorHandler` in the `ReactiveStreamsConsumer`
This commit is contained in:
Artem Bilan
2020-02-10 16:27:59 -05:00
parent 6591ce90f6
commit a8c471c061
1 changed files with 13 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -166,21 +166,25 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
this.subscription =
Flux.from(this.publisher)
.flatMap(this.reactiveMessageHandler::handleMessage)
.onErrorContinue(this::onError)
.onErrorContinue((ex, data) -> this.errorHandler.handleError(ex))
.subscribe();
}
else if (this.subscriber != null) {
Flux.from(this.publisher)
.doOnSubscribe((subs) -> this.subscription = subs::cancel)
.onErrorContinue(this::onError)
.subscribe(this.subscriber);
this.subscription =
Flux.from(this.publisher)
.doOnComplete(this.subscriber::onComplete)
.doOnSubscribe(this.subscriber::onSubscribe)
.subscribe((data) -> {
try {
this.subscriber.onNext(data);
}
catch (Exception ex) {
this.errorHandler.handleError(ex);
}
});
}
}
private void onError(Throwable ex, Object data) {
this.errorHandler.handleError(ex);
}
@Override
protected void doStop() {
if (this.subscription != null) {