I'm working on a Reactor project at the moment. One of the cases we have requires going out to a service for each element in a Flux, if any of those fetches fail, we want to fail the whole batch and prevent downstream events from occurring. Currently the code looks roughly like:
getFluxIdsFromDatabase()
.flatMap(this::fetchFromExternalService)
.collectList() // prevent downstream execution by having any errors blocked here
.flatMapMany(Flux::fromIterable) // back to regularly scheduled Flux if no errors
.concatMap(this::saveSequentiallyToDatabase)
This works... but it really seems to me like I should be able to do this without the switching to a Mono<List> just to halt execution on an error. When I try making the first line concatMap, I see saves to the database from some elements even though there was an error during the fetchFromExternalService. So is there some other Flux method I can call here to do the mapping that will halt everything on an error, or am I stuck with this collectList/flatMapMany(Flux::fromIterable) solution?