How to chain reactive operations after empty Mono without blocking?

Question

Basically what I am trying to achieve is to call a second repository (a ReactiveCrudRepository) or throw an exception, depending on the result of a call to a first repository.

My original idea looks like this:

/** Reactive with blocking code */
public Flux<SecondThing> getThings(String firstThingName) {
    FirstThing firstThing = firstRepo
        .findByName(firstThingName)
        // Warning: "Inappropriate blocking method call"
        .blockOptional()  // this fails in test-context
        .orElseThrow(() -> new FirstThingNotFound(firstThingName));

    return secondRepo.findAllByFirstThingId(firstThing.getId());
}

Which would correspond to the following non-reactive approach:

/** Non-reactive */
public List<SecondThing> getThings(String firstThingName) {
    FirstThing firstThing = firstRepo
        .findByName(firstThingName)
        .orElseThrow(() -> new FirstThingNotFound(firstThingName));

    return secondRepo.findAllByFirstThingId(firstThing.getId());
}

I haven't found a way to do this in a reactive non-blocking way. All I need is to throw an error if an empty Mono comes out of the first call, and continue the pipeline if not empty; but I could not seem to use onErrorStop or doOnError correctly here, and map does not help as it skips the empty Mono.

What I have is a workaround if I use id instead of name, but I'm not quite satisfied with it as it shows a different behaviour in the case where is an instance of FirstThing but no SecondThing linked to it:

/** Reactive workaround 1 */
public Flux<SecondThing> getThings(Long firstThingId) {
    return secondRepo
        .findAllByFirstThingId(firstThingId)
        .switchIfEmpty(
            Flux.error(() -> new FirstThingNotFound(firstThingName))
        );
}

Another workaround I've found is the following, that replaces the empty Mono with a null value, but it doesn't look right and throws a warning too:

/** Reactive workaround 2 */
public Flux<SecondThing> getThings(String firstThingName) {
    return firstRepo
        .findByName(firstThingName)
        // Warning: "Passing 'null' argument to parameter annotated as @NotNull"
        .defaultIfEmpty(null)
        .flatMapMany(
            firstThing -> secondRepo.findAllByFirstThingId(firstThing.getId()
        )
        .onErrorMap(
            NullPointerException.class, e -> new FirstThingNotFound(firstThingName)
        );
}

What is the correct way to chain the calls to both repositories so that the presence or absence of a FirstThing with the requested firstThingName conditions the call to the second repo?

Didn't look too closely but be aware that a reactive flow doesn't continue if it is empty. So, it seems to me you could just call get firstThingName and if it comes back empty it won't make the call to getFirstThing so is there a problem? — K.Nicholas
@K.Nicholas the problem is, I'd like to throw an exception when no FirstThing is found. So I guess one could rephrase my question to: "How to map to error from empty Mono?" — AdrienW
Well, if the firstRepo returns an Optional then you can do 'orElseThrow' . I think that's the cleanest answer. .map(op->op.orElseThrow(()->new IllegalArgumentException("name not found"))) — K.Nicholas
@K.Nicholas Agreed, but the ReactiveCrudRepository returns a Mono and converting this to an Optional requires a blocking operation, which I'm trying to avoid. What you propose would need a Mono<Optional<FirstThing>> to work. Not possible in my case as the repository class is auto-generated, but I'll save the idea for som other time. — AdrienW

AdrienW AdrienW · Accepted Answer · 2021-02-05T15:51:50

I found a solution so simple, that I could be ashamed not to have found it earlier:

public Flux<SecondThing> getThings(String firstThingName) {
    return firstRepo
        .findByName(firstThingName)
        .switchIfEmpty(Mono.error(() -> new FirstThingNotFound(firstThingName)))
        .flatMapMany(
            firstThing -> secondRepo.findAllByFirstThingId(firstThing.getId()
        );
}

The trick is that switchIfEmpty does not force you to pick a "valid" replacement value, so it is possible to use a Mono.error to propagate the right exception directly.

How to chain reactive operations after empty Mono without blocking?

1 Answers