Combination of two Mono with condition

Question

I want to combine result form two Mono based on some condition. Both Mono are results of WebClient calls:

• The first one is a single call expecting fast response.
• The second one is a combination of several calls with a slow response.

The idea to "cancel" the second Mono if result from the first one satisfies some condition to save time and avoid unnecessary network calls. If the first's Mono result is not enough zip it with the second Mono.

A Kotlin code sample to explain my idea:

fun getResult(): Mono<Result> {

    val trivialResultMono: Mono<Result> = webClient.getResult()

    val nonTrivialResultMono: Mono<Result> = webClient
            .getResult()
            .flatMap { webClient.getResult1(it) }
            .flatMap { webClient.getResult2(it) }
            .flatMap { webClient.getResult2(it) }

    //here I need to check if trivial result satisfies some condition,
    //for example trivialResult.size > 5 if it's true I just return
    //trivialResultMono from getResult() function, 
    //it it's false something like this:
    return Mono.zip(trivialResultMono, nonTrivialResultMono) { trivialResult, nonTrivialResult ->
        trivialResult + nonTrivialResult
    }

}

UPDATE:

To be more clear let's say that trivialResult comes in 1 second, nonTrivialResult in 2 seconds. I want to get my final result in 1 second in case of trivialResult.size > 5 and in 2 seconds otherwise.

Using just Mono.zip(trivialResultMono, nonTrivialResultMono) I will always get my final result in 2 seconds.

Using filter + switchIfEmpty it will take 1 second if trivialResult.size > 5 and 3 seconds otherwise. Please correct me if I wrong.

Sheelagh

Given that the first call is quick, can you not simply wait for it to complete before filtering out the inappropriate result and switching to the second call? – Feline

Let's assume that the second call is 1,5 times longer than the first. In this case this solution is not so optimal. – Sheelagh

Answer 1

You could filter your trivialResultMono and apply switchIfEmpty operator

return trivialResultMono
        .filter(trivialResult -> trivialResult.size > 5)
        .switchIfEmpty(Mono.zip(...))

Update for merge approach:

Mono<Result> zipResultMono = Mono.zip...

return Flux.merge(
        trivialResultMono.map(trivialResult -> Tuples.of(1, trivialResult)),
        zipResultMono.map(zipResult -> Tuples.of(2, zipResult)))
        .filter(tuple ->
                (tuple.getT1().equals(1) && tuple.getT2().size > 5) ||
                        tuple.getT1().equals(2))
        .next()
        .map(Tuple2::getT2);

You could skip converting to the Tuple2 if zipResult always has size more then 5

Hypercorrection

Looks like a solution but in case of trivialResult.size <= 5 time to get result will be longer than just zip without filter and switch. – Sheelagh

You could use trivialResult.cache(). Duration difference shouldn't be critical for almost all systems. – Hypercorrection

I mean in this case requests for trivial and non trivial results will be executed not concurrently but in serial way. – Sheelagh

In such case you could use Flux.merge then filter and take first result. I have updated answer. – Hypercorrection

Answer 2

You can achieve this with flatMap and map:

trivial.flatMap(trivialResult -> {
  if (trivialResult.size > 5) {
    return Mono.just(trivialResult);
  } else {
    return nonTrivial.map(nonTrivialResult -> trivialResult + nonTrivialResult);
  }
});

Horwath