Combining items from streams

Learn how to combine items from different streams.

Combining items from various streams is an essential pattern in Reactive Programming. It associates the emitted items from multiple streams and emits an aggregate. The downstream receives this aggregates and can handle it smoothly.

There are plenty of use cases, such as executing two tasks concurrently and waiting for both completions, getting the last items from different streams to build an always up-to-date view, and so on.

Combining Unis

Imagine that you have two asynchronous operations to perform like 2 HTTP requests. You want to send these requests and be notified when both have completed with their responses ready to be consumed.

Of course, you could send the first request, wait for the response, and then send the second request. If both requests are independent, we can do something better: send both concurrently and await for both completions!

How can you achieve this with Mutiny?

First, each request is a Uni, so we have:

Uni<Response> uniA = invokeHttpServiceA();
Uni<Response> uniB = invokeHttpServiceB();

Then, we want to combine both responses:

Uni<Tuple2<Response, Response>> responses = Uni.combine()
        .all().unis(uniA, uniB).asTuple();

This code creates a new Uni produced by combining uniA and uniB. The responses are aggregated inside a Tuple:

Uni.combine().all().unis(uniA, uniB).asTuple()
        .subscribe().with(tuple -> {
    System.out.println("Response from A: " + tuple.getItem1());
    System.out.println("Response from B: " + tuple.getItem2());
});

The tuple aggregates the responses in the same order as the Uni sequence.

If one of the Uni fails, so does the combination and you recieve the failure:

Uni<Response> uniA = invokeHttpServiceA();
Uni<Response> uniB = invokeHttpServiceB();
Uni<Tuple2<Response, Response>> responses = Uni.combine()
        .all().unis(uniA, uniB).asTuple();
Uni.combine().all().unis(uniA, uniB).asTuple()
        .subscribe().with(tuple -> {
    System.out.println("Response from A: " + tuple.getItem1());
    System.out.println("Response from B: " + tuple.getItem2());
});

Using tuples is convenient but only works if you have less than 10 Uni objects. If you want another structure or deal with more than 10 Uni objects then use combineWith:

Uni<Map<String, Response>> uni = Uni.combine()
        .all().unis(uniA, uniB).combinedWith(
                listOfResponses -> {
                    Map<String, Response> map = new LinkedHashMap<>();
                    map.put("A", (Response) listOfResponses.get(0));
                    map.put("B", (Response) listOfResponses.get(1));
                    return map;
                }
        );

Combining Multis

Combining Multis consists of associating items from different stream per index:

It associates the first items from the combined streams, then the second items:

Multi<Tuple2<A, B>> combined = Multi.createBy().combining()
        .streams(multiA, multiB).asTuple();

As for Uni, you can aggregate the item into tuples (up to 9 items) or combine with a combinator function:

Multi.createBy().combining()
        .streams(multiA, multiB).using(list -> combineItems(list))
        .subscribe().with(x -> {
            // do something with the combined items
        });

If one of the streams fails, the combined stream propagates the failure and stops the emission. The combined stream completes as soon as one of the observed stream sends the completion event.

If one of the observed streams never emits any item then the combined stream will not emit anything.

Combining the last items of Multis

It can be useful to combine multiple Multi streams and receive the latest items from each stream on every emission:

Combining Multis to always gets the latest items from every stream

This is achieved using latest():

Multi<Tuple2<A, B>> multi1 = Multi.createBy().combining()
        .streams(multiA, multiB)
        .latestItems().asTuple();

// or

Multi<String> multi2 = Multi.createBy().combining()
        .streams(multiA, multiB)
        .latestItems().using(list -> combineItems(list));