Mutiny asynchronous stream processing: efficient concurrent processing of elements in Uni

1. Problem background and common misunderstandings

In reactive programming, we often encounter the need to handle asynchronous operations involving multiple elements. For example, given a Uni>, we want to perform a time-consuming asynchronous task on each string in the list and eventually collect or process the results of all tasks.

A common attempt is to use map to convert List to List>, and then merge these Unis via Uni.join().all(unis).andCollectFailures(). However, this method may not achieve the expected concurrent processing effect, or in a short-life cycle program (such as a unit test), the asynchronous task will be terminated before being completed due to the premature exit of the main thread, thus giving the illusion that "only the first element has been processed."

The core of the problem is that Uni> itself represents a single value stream, and its value is a complete list. If you want to operate asynchronously on each element in the list and treat it as an independent reactive event, you need to "unroll" the list into a stream that can be processed one by one. Mutiny provides the Multi type to handle streams of zero to N elements, which is the key to solving such problems.

2. Mutiny asynchronous stream processing core: Uni and Multi

Mutiny is a widely used reactive programming library in frameworks such as Quarkus. It provides two core types:

• Uni : Represents an asynchronous operation that will eventually emit 0 or 1 elements, or a failure event.
• Multi : Represents an asynchronous operation stream that can emit 0 to N elements, or a failure event, and will eventually emit a completion event.

To achieve asynchronous and concurrent processing of each element in Uni>, we need to first convert Uni> into a Multi, so that each string in the list becomes an independent event in the Multi stream. We can then apply an asynchronous transformation to each event in this Multi stream.

3. Solution 1: Handle asynchronous streams gracefully in a test environment (combined with Vert.x Unit)

In unit testing or scenarios that require non-blocking waiting for all asynchronous operations to be completed, we can use the Multi feature and the onTermination().invoke() callback to ensure that all tasks are completed. The following example incorporates a Vert.x Unit, which provides an Async mechanism to manage the lifecycle of asynchronous tests.

 import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import io.vertx.core.Vertx;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;

import java.time.Duration;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@ExtendWith(VertxExtension.class)
public class AsyncListProcessingTest {

    // Simulate an asynchronous operation and return a Uni
    private Uni<string> processItemAsync(String item, Random random) {
        final int duration = (random.nextInt(5) 1) * 1000; // Random delay of 1-5 seconds System.out.println("Starting process for: " item ", duration: " duration "ms");
        return Uni.createFrom().item(item)
                .onItem().delayIt().by(Duration.ofMillis(duration))
                .invoke(() -&gt; System.out.println("Finished process for: " item));
    }

    @Test
    public void testAsyncProcessingWithVertxUnit(VertxTestContext context) {
        Random random = new Random();
        // The Async object of Vert.x Unit, used to notify the test framework when the asynchronous operation is completed context.verify(() -&gt; { // Ensure that Uni.createFrom() is executed in the context of VertxTestContext
                    .item(List.of("a", "b", "c")) // Initial Uni<list>&gt;
                    // 1. Convert Uni<list>&gt; to Multi<string>
                    .onItem().transformToMulti(Multi.createFrom()::iterable)
                    // 2. Apply an asynchronous transformation to each element in the Multi and merge the results back into the Multi
                    .onItem().transformToUniAndMerge(s -&gt; processItemAsync(s, random))
                    // 3. Subscribe to the Multi stream and process each completed element.subscribe()
                    .with(
                            s -&gt; System.out.println("Printing result: " s), // Successfully process each element context::failNow, // Any error causes the stream to fail context::completeNow // The stream is completed, notifying VertxTestContext that the test is over);
        });
    }
}</string></list></list></string>

Code explanation:

1. Uni.createFrom().item(List.of("a", "b", "c")) : Create an initial Uni containing a list of strings.
2. .onItem().transformToMulti(Multi.createFrom()::iterable) : This is the key step to convert Uni> to Multi. It expands the contents of the list emitted by Uni so that each element in the list becomes a new Multi event.
3. .onItem().transformToUniAndMerge(s -> processItemAsync(s, random)) : This is the core of implementing concurrent asynchronous processing.
   • transformToUni: For each element s in Multi, the processItemAsync(s, random) method is called, which returns a Uni.
   • andMerge: Mutiny will concurrently subscribe to and execute these Unis created by transformToUni. When any Uni completes, its results are immediately merged into the output Multi stream. This means that the order of the results may not be the same as the original list, but instead depends on which asynchronous operation completes first.
4. .subscribe().with(...) : Subscribe to the final Multi stream.
   • The first Lambda expression handles each successful result emitted in the Multi.
   • The second Lambda expression handles any errors in the stream.
   • The third Lambda expression is called when the stream completes successfully, here we use context::completeNow to notify Vert.x Unit that the test has successfully completed all asynchronous operations.

4. Solution 2: Blocking wait for all asynchronous results

In some scenarios, such as command line tools or when we need to wait for all asynchronous operations to complete before continuing the main program execution, we can choose to block the current thread until all results are collected.

 import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;

import java.time.Duration;
import java.util.List;
import java.util.Random;

public class BlockingAsyncListProcessing {

    private static Uni<string> processItemAsync(String item, Random random) {
        final int duration = (random.nextInt(5) 1) * 1000; // Random delay of 1-5 seconds return Uni.createFrom().item(item)
                .onItem().delayIt().by(Duration.ofMillis(duration))
                .invoke(() -&gt; System.out.println("Letter: " item ", duration in ms: " duration));
    }

    public static void main(String[] args) {
        Random random = new Random();

        System.out.println("Starting blocking asynchronous processing...");

        List<string> results = Uni.createFrom()
                .item(List.of("a", "b", "c")) // Initial Uni<list>&gt;
                // 1. Convert Uni<list>&gt; to Multi<string>
                .onItem().transformToMulti(Multi.createFrom()::iterable)
                // 2. Apply an asynchronous transformation to each element in the Multi and merge the results back into the Multi
                .onItem().transformToUniAndMerge(s -&gt; processItemAsync(s, random))
                // 3. Optional: handle each completed element.onItem().invoke(s -&gt; System.out.println("Printing collected item: " s))
                // 4. Collect all elements in Multi into a list.collect().asList()
                // 5. Block the current thread until Uni<list>&gt; completes and returns the result.await().indefinitely();

        System.out.println("All items processed. Collected results: " results);
    }
}</list></string></list></list></string></string>

Code explanation:

1. The first two steps are the same as solution one : convert Uni> to Multi, and then use onItem().transformToUniAndMerge() to process each element concurrently.
2. .collect().asList() : This operator collects all emitted elements in the Multi stream into a List, and finally returns a Uni>. This Uni will emit a list containing all collected elements when the source Multi completes.
3. .await().indefinitely() : This is a blocking operation. It blocks the current thread until the upstream Uni> emits its result (i.e. all asynchronous operations complete and the results are collected into the list). indefinitely() means waiting indefinitely.

5. Precautions and best practices

• Non-blocking first : Use non-blocking reactive mode as much as possible (such as solution 1). The await() operation blocks the current thread and should be used with caution in production environments, especially in I/O-intensive or web applications, as it may cause thread starvation and performance issues. It is more suitable for startup code, testing, or specific scenarios where you need to wait synchronously for all asynchronous tasks to complete.
• Error handling : There should be complete error handling logic in the Uni created inside transformToUniAndMerge and in the final subscribe().with() method. Mutiny provides a wealth of error handling operators, such as onFailure().recoverWith(), onFailure().retry(), etc.
• Concurrency : transformToUniAndMerge will process tasks concurrently, but the actual concurrency may be limited by the underlying thread pool configuration, system resources, and the implementation of specific asynchronous operations. If you need fine control over concurrency, consider using the transformToUniAndMerge(concurrency, ...) variant.
• Order guarantee : transformToUniAndMerge does not guarantee that the order of the results is consistent with the order of the original list. If you need to maintain order, consider using transformToUniAndConcatenate or sorting manually after collection.
• Resource management : Ensure that any external resources (such as database connections, file handles) used in asynchronous operations are properly managed and released.

6. Summary

Through Mutiny's Multi type and onItem().transformToUniAndMerge() operator, we can effectively convert each element in Uni> into an independent asynchronous task and process it concurrently. Depending on the application scenario, we can choose the non-blocking subscription mode (suitable for reactive systems and testing) or the blocking await() mode (suitable for specific scenarios that require synchronous waiting for results). Understanding and correctly using these Mutiny operators is key to building efficient and robust reactive applications.