Project Reactor conditional execution

Question

I have an object to save (to MongoDB), but before it I need to check if some conditions are true.

Object contains IDs to other objects. It looks like

"object": {
   "id": "123",
   "subobject1": { "id": "1" },
   "subobject2": { "id": "2" }
}

Subobjects contain only id, other info is located in other collection, so I have to check is the information exist.

In block-style I can do something like

    if (!languageRepository.exists(Example.of(wordSet.getNativeLanguage())).block()) {
        throw new RuntimeException("Native language doesn't exist");
    }

    if (!languageRepository.exists(Example.of(wordSet.getTargetLanguage())).block()) {
        throw new RuntimeException("Target language doesn't exist");
    }

and only then I can save my object

return wordSetRepository.save(wordSet);

How can I do it in "reactive" style without blocking?

Answer 1

If you want to propagate distinct errors for the native vs target language error cases, you'll need to perform async filtering inside a flatMap:

objectFlux.flatMap(o ->
    Mono.just(o)
        .filterWhen(languageRepository.exists(...)) //native
        .switchIfEmpty(Mono.error(new RuntimeException("Native language doesn't exist"))
        .filterWhen(languageRepository.exists(...)) //target
        .switchIfEmpty(Mono.error(new RuntimeException("Target language doesn't exist"))
    )
    .flatMap(wordSetRepository::save);

The async filtering inside the flatMap ensures that if the test doesn't pass, the inner sequence is empty. This in turn allows us to detect the case and propagate the adequate error. If both tests pass, the original o is propagated in the main sequence.

The second flatMap takes it from there, only receiving the elements that passed both filters and saving them in DB.

Note that the first element to not pass the filters will interrupt the whole sequence (but it was the same in the blocking code since an exception was thrown).