FluxMessageChannel: try.catch not onErrorContinue

It turns out that some upstream fluxes may come with the
`onErrorResume()` logic.
The `onErrorContinue()` here downstream in the `FluxMessageChannel.java`
eliminates an `onErrorResume()` purpose.

* Change the logic to `try..catch` in the `doOnNext()` instead and let
that upstream `onErrorResume()` to do its job
This commit is contained in:
Artem Bilan
2020-01-22 13:15:56 -05:00
parent a62a572a6c
commit 9b7b0d91ac

View File

@@ -78,8 +78,14 @@ public class FluxMessageChannel extends AbstractMessageChannel
Flux.from(publisher)
.delaySubscription(this.subscribedSignal.filter(Boolean::booleanValue).next())
.publishOn(Schedulers.boundedElastic())
.doOnNext(this::send)
.onErrorContinue((ex, message) -> logger.warn("Error during processing event: " + message, ex))
.doOnNext((message) -> {
try {
send(message);
}
catch (Exception ex) {
logger.warn("Error during processing event: " + message, ex);
}
})
.subscribe());
}