Combining Publishers in Project Reactor

1. Overview

In this article, we’ll take a look at various ways of combining Publishers in Project Reactor.

2. Maven Dependencies

Let’s set up our example with the Project Reactor dependencies:

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.6.0</version>
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <version>3.6.0</version>
    <scope>test</scope>
</dependency>

3. Combining Publishers

Given a scenario when one has to work with Flux<T> or Mono<T>, there are different ways to combine streams.

Let’s create a few examples to illustrate the usage of static methods in the Flux<T> class such as concat, concatWith, merge, zip and combineLatest.

Our examples will make use of two publishers of type Flux<Integer>, namely evenNumbers, which is a Flux of Integer and holds a sequence of even numbers starting with 1 (min variable) and limited by 5 (max variable).

We’ll create oddNumbers, also a Flux of type Integer of odd numbers:

Flux<Integer> evenNumbers = Flux
  .range(min, max)
  .filter(x -> x % 2 == 0); // i.e. 2, 4

Flux<Integer> oddNumbers = Flux
  .range(min, max)
  .filter(x -> x % 2 > 0);  // ie. 1, 3, 5

3.1. concat()

The concat method executes a concatenation of the inputs, forwarding elements emitted by the sources downstream.

The concatenation is achieved by sequentially subscribing to the first source then waiting for it to complete before subscribing to the next, and so on until the last source completes. Any error interrupts the sequence immediately and is forwarded downstream.

Here is a quick example:

@Test
public void givenFluxes_whenConcatIsInvoked_thenConcat() {
    Flux<Integer> fluxOfIntegers = Flux.concat(
      evenNumbers, 
      oddNumbers);
        
    StepVerifier.create(fluxOfIntegers)
      .expectNext(2)
      .expectNext(4)
      .expectNext(1)
      .expectNext(3)
      .expectNext(5)
      .expectComplete()
      .verify();
}

3.2. concatWith()

Using the static method concatWith, we’ll produce a concatenation of two sources of type Flux<T> as a result:

@Test
public void givenFluxes_whenConcatWithIsInvoked_thenConcatWith() {
    Flux<Integer> fluxOfIntegers = evenNumbers.concatWith(oddNumbers);
        
    // same stepVerifier as in the concat example above
}

3.3. combineLatest()

The Flux static method combineLatest will generate data provided by the combination of the most recently published value from each of the Publisher sources.

Here’s an example of the usage of this method with two Publisher sources and a BiFunction as parameters:

@Test
public void givenFluxes_whenCombineLatestIsInvoked_thenCombineLatest() {
    Flux<Integer> fluxOfIntegers = Flux.combineLatest(
      evenNumbers, 
      oddNumbers, 
      (a, b) -> a + b);

    StepVerifier.create(fluxOfIntegers)
      .expectNext(5) // 4 + 1
      .expectNext(7) // 4 + 3
      .expectNext(9) // 4 + 5
      .expectComplete()
      .verify();
}

We can see here that the function combineLatest applied the function “a + b” using the latest element of evenNumbers (4) and the elements of oddNumbers (1,3,5), thus generating the sequence 5,7,9.

3.4. merge()

The merge function executes a merging of the data from Publisher sequences contained in an array into an interleaved merged sequence:

@Test
public void givenFluxes_whenMergeIsInvoked_thenMerge() {
    Flux<Integer> fluxOfIntegers = Flux.merge(
      evenNumbers, 
      oddNumbers);
        
    StepVerifier.create(fluxOfIntegers)
      .expectNext(2)
      .expectNext(4)
      .expectNext(1)
      .expectNext(3)
      .expectNext(5)
      .expectComplete()
      .verify();
}

An interesting thing to note is that, opposed to concat (lazy subscription), the sources are subscribed eagerly.

Here, we can see a different outcome of the merge function if we insert a delay between the elements of the Publishers:

@Test
public void givenFluxes_whenMergeWithDelayedElementsIsInvoked_thenMergeWithDelayedElements() {
    Flux<Integer> fluxOfIntegers = Flux.merge(
      evenNumbers.delayElements(Duration.ofMillis(500L)), 
      oddNumbers.delayElements(Duration.ofMillis(300L)));
        
    StepVerifier.create(fluxOfIntegers)
      .expectNext(1)
      .expectNext(2)
      .expectNext(3)
      .expectNext(5)
      .expectNext(4)
      .expectComplete()
      .verify();
}

3.5. mergeSequential()

The mergeSequential method merges data from Publisher sequences provided in an array into an ordered merged sequence.

Unlike concat, sources are subscribed to eagerly.

Also, unlike merge, their emitted values are merged into the final sequence in subscription order:

@Test
public void testMergeSequential() {
    Flux<Integer> fluxOfIntegers = Flux.mergeSequential(
      evenNumbers, 
      oddNumbers);
        
    StepVerifier.create(fluxOfIntegers)
      .expectNext(2)
      .expectNext(4)
      .expectNext(1)
      .expectNext(3)
      .expectNext(5)
      .expectComplete()
      .verify();
}

3.6. mergeDelayError()

The mergeDelayError merges data from Publisher sequences contained in an array into an interleaved merged sequence.

Unlike concat, sources are subscribed to eagerly.

This variant of the static merge method will delay any error until after the rest of the merge backlog has been processed.

Here is an example of mergeDelayError:

@Test
public void givenFluxes_whenMergeWithDelayedElementsIsInvoked_thenMergeWithDelayedElements() {
    Flux<Integer> fluxOfIntegers = Flux.mergeDelayError(1, 
      evenNumbers.delayElements(Duration.ofMillis(500L)), 
      oddNumbers.delayElements(Duration.ofMillis(300L)));
        
    StepVerifier.create(fluxOfIntegers)
      .expectNext(1)
      .expectNext(2)
      .expectNext(3)
      .expectNext(5)
      .expectNext(4)
      .expectComplete()
      .verify();
}

3.7. mergeWith()

The static method mergeWith merges data from this Flux and a Publisher into an interleaved merged sequence.

Again, unlike concat, inner sources are subscribed to eagerly:

@Test
public void givenFluxes_whenMergeWithIsInvoked_thenMergeWith() {
    Flux<Integer> fluxOfIntegers = evenNumbers.mergeWith(oddNumbers);
        
    // same StepVerifier as in "3.4. Merge"
    StepVerifier.create(fluxOfIntegers)
      .expectNext(2)
      .expectNext(4)
      .expectNext(1)
      .expectNext(3)
      .expectNext(5)
      .expectComplete()
      .verify();
}

3.8. zip()

The static method zip agglutinates multiple sources together, i.e., waits for all the sources to emit one element and combines these elements into an output value (constructed by the provided combinator function).

The operator will continue doing so until any of the sources completes:

@Test
public void givenFluxes_whenZipIsInvoked_thenZip() {
    Flux<Integer> fluxOfIntegers = Flux.zip(
      evenNumbers, 
      oddNumbers, 
      (a, b) -> a + b);
        
    StepVerifier.create(fluxOfIntegers)
      .expectNext(3) // 2 + 1
      .expectNext(7) // 4 + 3
      .expectComplete()
      .verify();
}

As there is no element left from evenNumbers to pair up, the element 5 from oddNumbers Publisher is ignored.

3.9. zipWith()

The zipWith executes the same method that zip does, but only with two Publishers:

@Test
public void givenFluxes_whenZipWithIsInvoked_thenZipWith() {
    Flux<Integer> fluxOfIntegers = evenNumbers
     .zipWith(oddNumbers, (a, b) -> a * b);
        
    StepVerifier.create(fluxOfIntegers)
      .expectNext(2)  // 2 * 1
      .expectNext(12) // 4 * 3
      .expectComplete()
      .verify();
}

4. Conclusion

In this quick tutorial, we’ve discovered multiple ways of combining Publishers.