Conditional Statements in Spring WebFlux Reactive Flow

1. Overview

Using conditional statements in a Spring WebFlux reactive flow allows for dynamic decision-making while processing reactive streams. Unlike an imperative approach, conditional logic in a reactive approach is not limited to if-else statements. Instead, we can use a variety of operators, such as map(), filter(), swithIfEmpty(), and so on, to introduce conditional flow without blocking the stream.

In this article, we’ll explore different approaches to using conditional statements with Spring WebFlux. Unless specified explicitly, each approach will apply to both Mono and Flux.

2. Using Conditional Constructs with map()

We can use the map() operator to transform individual elements of the stream. Further, we can use if-else statements within the mapper to modify elements conditionally.

Let’s define a Flux named oddEvenFlux and label each of its elements as “Even” or “Odd” using the map() operator:

Flux<String> oddEvenFlux = Flux.just(1, 2, 3, 4, 5, 6)
  .map(num -> {
    if (num % 2 == 0) {
      return "Even";
    } else {
      return "Odd";
    }
  });

We should note that map() is synchronous, and it applies the transformation function immediately after an item is emitted.

Next, let’s use the StepVerifier to test the behavior of our reactive stream and confirm the conditional labeling of each item:

StepVerifier.create(oddEvenFlux)
  .expectNext("Odd")
  .expectNext("Even")
  .expectNext("Odd")
  .expectNext("Even")
  .expectNext("Odd")
  .expectNext("Even")
  .verifyComplete();

As expected, each number is labeled according to its parity.

3. Using filter()

We can use the filter() operator to filter out data using a predicate, ensuring that the downstream operators receive only the relevant data.

Let’s create a new Flux named evenNumbersFlux from a stream of numbers:

Flux<Integer> evenNumbersFlux = Flux.just(1, 2, 3, 4, 5, 6)
  .filter(num -> num % 2 == 0);

Here, we’ve added a predicate for the filter() operator to determine if the number is even.

Now, let’s verify that evenNumbersFlux allows only even numbers to pass downstream:

StepVerifier.create(evenNumbersFlux)
  .expectNext(2)
  .expectNext(4)
  .expectNext(6)
  .verifyComplete();

Great! It works as expected.

4. Using switchIfEmpty() and defaultIfEmpty()

In this section, we’ll learn about two useful operators that enable conditional data flow when the underlying flux doesn’t emit any items.

4.1. With switchIfEmpty()

When an underlying flux doesn’t publish any items, we might want to switch to an alternative stream. In such a scenario, we can supply an alternative publisher via the switchIfEmpty() operator.

Let’s say we’ve got a flux of words chained with a filter() operator that only allows words that have a length of two or more characters:

Flux<String> flux = Flux.just("A", "B", "C", "D", "E")
  .filter(word -> word.length() >= 2);

Naturally, when none of the words meet the filter criteria, the flux won’t emit any items.

Now, let’s supply an alternative flux via the switchIfEmpty() operator:

flux = flux.switchIfEmpty(Flux.defer(() -> Flux.just("AA", "BB", "CC")));

We’ve used the Flux.defer() method to ensure that the alternative flux is created only when the upstream flux doesn’t produce any items.

Lastly, let’s verify that the resultant flux produces all items from the alternate source:

StepVerifier.create(flux)
  .expectNext("AA")
  .expectNext("BB")
  .expectNext("CC")
  .verifyComplete();

The result looks correct.

4.2. With defaultIfEmpty()

Alternatively, we can use the defaultIfEmpty() operator to supply a fallback value instead of an alternative publisher when the upstream flux doesn’t emit any item:

flux = flux.defaultIfEmpty("No words found!");

Another key difference between using the switchIfEmpty() and defaultIfEmpty() is that we’re limited to using a single default value with the latter.

Now, let’s verify the conditional flow of our reactive stream:

StepVerifier.create(flux)
  .expectNext("No words found!")
  .verifyComplete();

We’ve got this one right.

5. Using flatMap()

We can use the flatMap() operator to create multiple conditional branches in our reactive stream while maintaining a non-blocking, asynchronous flow.

Let’s take a look at a Flux created from words and changed with two flatMap() operators:

Flux<String> flux = Flux.just("A", "B", "C")
  .flatMap(word -> {
    if (word.startsWith("A")) {
      return Flux.just(word + "1", word + "2", word + "3");
    } else {
      return Flux.just(word);
    }
  })
  .flatMap(word -> {
    if (word.startsWith("B")) {
      return Flux.just(word + "1", word + "2");
    } else {
      return Flux.just(word);
    }
  });

We’ve created a dynamic branching by adding two stages of conditional transformation, thereby providing multiple logical paths to each item of the reactive stream.

Now, it’s time to verify the conditional flow of our reactive stream:

StepVerifier.create(flux)
  .expectNext("A1")
  .expectNext("A2")
  .expectNext("A3")
  .expectNext("B1")
  .expectNext("B2")
  .expectNext("C")
  .verifyComplete();

Fantastic! It looks like we nailed this one. Further, we can use flatMapMany() for Mono publishers for a similar use case.

6. Using Side-Effect Operators

In this section, we’ll explore how to perform condition-based synchronous actions while processing a reactive stream.

6.1. With doOnNext()

We can use the doOnNext() operator to execute a side-effect operation synchronously for each item of the reactive stream.

Let’s start by defining the evenCounter variable to track the count of even numbers in our reactive stream:

AtomicInteger evenCounter = new AtomicInteger(0);

Now, let’s create a flux of integers and chain it together with a doOnNext() operator to increment the count of even numbers:

Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6)
  .doOnNext(num -> {
  if (num % 2 == 0) {
    evenCounter.incrementAndGet();
  }
  });

We’ve added the action within an if-block, thereby enabling a conditional increment of the counter.

Next, we must verify the logic and state of evenCounter after each item from the reactive stream is processed:

StepVerifier.create(flux)
  .expectNextMatches(num -> num == 1 && evenCounter.get() == 0)
  .expectNextMatches(num -> num == 2 && evenCounter.get() == 1)
  .expectNextMatches(num -> num == 3 && evenCounter.get() == 1)
  .expectNextMatches(num -> num == 4 && evenCounter.get() == 2)
  .expectNextMatches(num -> num == 5 && evenCounter.get() == 2)
  .expectNextMatches(num -> num == 6 && evenCounter.get() == 3)
  .verifyComplete();

Great! We’ve got the expected results.

6.2. With doOnComplete()

Similarly, we can also associate actions based on the condition of receiving a signal from the reactive stream, such as the complete signal sent after it has published all its items.

Let’s start by initializing the done flag:

AtomicBoolean done = new AtomicBoolean(false);

Now, let’s define a flux of integers and add an action of setting the done flag to true using the doOnComplete() operator:

Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6)
  .doOnComplete(() -> done.set(true));

It’s important to note that the complete signal is sent only once, so the side-effect action will be triggered at most once.

Further, let’s verify the conditional execution of the side effect by validating the done flag at various steps:

StepVerifier.create(flux)
  .expectNextMatches(num -> num == 1 && !done.get())
  .expectNextMatches(num -> num == 2 && !done.get())
  .expectNextMatches(num -> num == 3 && !done.get())
  .expectNextMatches(num -> num == 4 && !done.get())
  .expectNextMatches(num -> num == 5 && !done.get())
  .expectNextMatches(num -> num == 6 && !done.get())
  .then(() -> Assertions.assertTrue(done.get()))
  .expectComplete()
  .verify();

Perfect! We can see that the done flag was set to true only after all the items were emitted successfully. However, it’s important to note that doOnComplete() applies only to Flux publishers, and we must use doOnSuccess() for Mono publishers.

7. Using firstOnValue()

Sometimes, we might have multiple sources to collect data, but each could have a different latency. From a performance point of view, it’s best to use the value from the source with the least latency. For such conditional data access, we can use the firstOnValue() operator.

First, let’s define two sources, namely, source1 and source2, with latency of 200 ms and 10 ms, respectively:

Mono<String[]> source1 = Mono.defer(() -> Mono.just(new String[] { "val", "source1" })
  .delayElement(Duration.ofMillis(200)));
Mono<String[]> source2 = Mono.defer(() -> Mono.just(new String[] { "val", "source2" })
  .delayElement(Duration.ofMillis(10)));

Next, let’s use the firstWithValue() operator with the two sources and rely on the framework’s conditional logic to handle data access:

Mono<String[]> mono = Mono.firstWithValue(source1, source2);

Finally, let’s verify the outcome by comparing the emitted item against the data from the source with lower latency:

StepVerifier.create(mono)
  .expectNextMatches(item -> "val".equals(item[0]) && "source2".equals(item[1]))
  .verifyComplete();

Excellent! We’ve got this one right. Further, it’s important to note that firstWithValue() is available only for Mono publishers.

8. Using zip() and zipWhen()

In this section, let’s learn how to leverage the conditional flow using the zip() and zipWhen() operators.

8.1. With zip()

We can use the zip() operator to combine emissions from multiple sources. Further, we can use its combinator function to add conditional logic for data processing. Let’s see how we can use this to determine if values in a cache and database are inconsistent

First, let’s define the dataFromDB and dataFromCache publishers to simulate sources with varying latencies and values:

Mono<String> dataFromDB = Mono.defer(() -> Mono.just("db_val")
  .delayElement(Duration.ofMillis(200)));
Mono<String> dataFromCache = Mono.defer(() -> Mono.just("cache_val")
  .delayElement(Duration.ofMillis(10)));

Now, let’s zip them and use its combinator function to add the condition for determining if the cache is consistent:

Mono<String[]> mono = Mono.zip(dataFromDB, dataFromCache, 
  (dbValue, cacheValue) -> 
  new String[] { dbValue, dbValue.equals(cacheValue) ? "VALID_CACHE" : "INVALID_CACHE" });

Finally, let’s verify this simulation and validate that the cache is inconsistent because db_val is different from cache_val:

StepVerifier.create(mono)
  .expectNextMatches(item -> "db_val".equals(item[0]) && "INVALID_CACHE".equals(item[1]))
  .verifyComplete();

The result looks correct.

8.2. With zipWhen()

The zipWhen() operator is more suitable when we want to collect the emission from the second source only in the presence of an emission from the first source. Further, we can use the combinator function to add conditional logic to process our reactive stream.

Let’s say that we want to compute the age category of a user:

int userId = 1;
Mono<String> userAgeCategory = Mono.defer(() -> Mono.just(userId))
  .zipWhen(id -> Mono.defer(() -> Mono.just(20)), (id, age) -> age >= 18 ? "ADULT" : "KID");

We’ve simulated the scenario for a user with a valid user ID. So, we’re guaranteed an emission from the second publisher. Subsequently, we’ll get the user’s age category.

Now, let’s verify this scenario and confirm that a user with the age of 20 is categorized as an “ADULT”:

StepVerifier.create(userDetail)
  .expectNext("ADULT")
  .verifyComplete();

Next, let’s use Mono.empty() to simulate the categorization for a scenario where a valid user is not found:

Mono<String> noUserDetail = Mono.empty()
  .zipWhen(id -> Mono.just(20), (id, age) -> age >= 18 ? "ADULT" : "KID");

Lastly, we can confirm that there are no emissions in this case:

StepVerifier.create(noUserDetail)
  .verifyComplete();

Perfect! We got the expected results for both scenarios. Further, we need to note that zipWhen() is available only for Mono publishers.

9. Conclusion

In this article, we learned how to include conditional statements in Spring WebFlux. Further, we explored how conditional flow is facilitated by different operators, such as map(), flatMap(), zip(), firstOnValue(), switchIfEmpty(), defaultIfEmpty(), and more.