Concurrent asynchronous actions with Mutiny

[giscus[bot]]

Concurrent asynchronous actions with Mutiny

Quarkus: Supersonic Subatomic Java

https://quarkus.io/blog/mutiny-concurrent-uni/

[syr]

Hello Clement, thank you for the interesting example. How would it have to be changed to have the two requests to two different services independent, e.g. if call to remote service A fails, response from remote service B should still be processed and not blocked by the failing remote service A, thus like the opposite of combine somehow.
Appreciate your thoughts :)

[tmarwen]

@syr, you should be able to use onFailure operator on the expected-to-fail (or both) remote calls:

private static Uni<String> getProgrammingQuote(WebClient client) {
    return client.getAbs(PROGRAMMING_QUOTE)
            .as(BodyCodec.jsonObject())
            .send()
            .onFailure()
            .recoverWithItem("Fallback")
            .onItem().transform(r -> r.body().getString("en") + " (" + r.body().getString("author") + ")");
}

The final Uni would still be composed the same way.

[liudas-stonys]

Does this example really show anything, which suppose to be executed in PARALLEL? My logs are showing, that after Uni.combine() every call is being executed concurrently on the same thread.

[liudas-stonys]

I made it to work in parallel on multiple threads by subscribing to Uni without any combine(), join(), etc. 🥳

[DevEnol]

@liudas-stonys
I tried that example, and it work as expected.
In my case, instead of using programming quotes and Chuck Norris remote service. I made custom REST API endpoints, (also with Quarkus) that would deliberately delay the response.
The code then would look like this:

Uni<Tuple2<String, String>> tuple = Uni.combine().all()
    .unis(getDelay2Second(client), getDelay4Second(client), getDelay8Second(client))
    .asTuple();

When I measure the time, the process finished in 8 second, which is the longest response of each responses.

In a non-reactive approach (blocking), the process would be finished at around 2 + 4 + 8 = 14 second .

[liudas-stonys]

@DevEnol, example works, I agree. But it's being done concurrently on a single thread and not in parallel as Mr. Clement have written: user wanted to call two microservices in parallel. He did the same multiple times: in podcast, Stackoverflow, so it got me confused, because I had to do those API calls in parallel.
To do it in parallel I've used emitOn(FixedThreadPool) on Multi or runSubscriptionOn(FixedThreadPool) on Uni. It didn't work with combine() method thought. And my logic totally agrees, that combined calls should be done on a single thread. I would love to be proved wrong thought:)
Here are working examples:

Multi.createBy().repeating()
        .uni(() -> Uni.createFrom().item(callUrl("https://api.chucknorris.io/jokes/random")))
        .atMost(3)
        .emitOn(Executors.newFixedThreadPool(3))
        .subscribe().with(
            log::info,
            err -> log.info(Arrays.toString(err.getStackTrace())));

Multi.createFrom().items(
            Uni.createFrom().item(callUrl("https://api.chucknorris.io/jokes/random")),
            Uni.createFrom().item(callUrl("https://api.chucknorris.io/jokes/random")),
            Uni.createFrom().item(callUrl("https://api.chucknorris.io/jokes/random")))
        .subscribe().with(
            uni -> uni.runSubscriptionOn(Executors.newFixedThreadPool(3)).subscribe().with(
                log::info,
                err -> log.warning(err.getMessage())),
            err -> log.warning(err.getMessage())
        );

[DevEnol]

Cool, I might have similar use case in the future, thanks for sharing the code @liudas-stonys

[jordanhuaman]

How i can run this? when i try to run this inside a function I get an error like ERROR [io.sma.graphql] (vert.x-eventloop-thread-4) SRGQL012000: Data Fetching Error: java.lang.IllegalStateException: Illegal pop() with non-matching JdbcValuesSourceProcessingState::

 Uni<List<CertificationEntity>> cert = certificationRepository.findByIds(certIds);
            Uni<List<EducationEntity>> edu = educationRepository.findByIds(certIds);
            return Uni.combine().all().unis(cert, edu)
                    .asTuple()
                    .map(tuple -> {
                        CompleteApplicationProfile completeApplicationProfile = new CompleteApplicationProfile();
                        System.out.println(tuple.getItem2());
                        return completeApplicationProfile;
                    });