Using zipWhen() With Mono

1. Overview

In this tutorial, we’ll explore how we can use zipWhen() to combine the results of two or more Mono streams in a coordinated manner. We’ll start with a quick overview. Next, we’ll set up a simple example involving user data storage and email. We’ll show how zipWhen() enables us to orchestrate and coordinate multiple asynchronous operations in scenarios where we need to gather and process data from various sources concurrently.

2. What Is zipWhen()?

In Reactive Programming with Mono, zipWhen() is an operator that allows us to combine the results of two or more Mono streams in a coordinated manner. It’s commonly used when we have multiple asynchronous operations to be performed concurrently, and we need to combine their results into a single output. 

We start with two or more Mono streams that represent asynchronous operations. These Monos can emit different types of data, and they may or may not have dependencies on each other.

We then use zipWhen() to coordinate. We apply the zipWhen() operator to one of the Monos. This operator waits for the first Mono to emit a value and then uses that value to trigger the execution of other Monos. The result of zipWhen() is a new Mono that combines the results of all the Monos into a single data structure, typically a Tuple or an object that we define.

Finally, we can specify how we want to combine the results of the Monos. We can use the combined values to create a new object, perform calculations, or construct a meaningful response.

3. Example Setup

Let’s set up a simple example consisting of three simplified services: UserService, EmailService, and DataBaseService. Each one of them produces data in the form of a Mono of different types. We want to combine all of the data in a single response and return it to the calling client. Let’s set up the appropriate POM dependencies first.

3.1. Dependencies

Let’s set up the required dependencies first. We’ll require spring-boot-starter-webflux and reactor-test:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId
    <artifactId>reactor-test</artifactId>
    <scope>test</scope>
</dependency>

3.2. Setting up UserService

Let’s start by introducing the User Service:

public class UserService {
    public Mono<User> getUser(String userId) {
        return Mono.just(new User(userId, "John Stewart"));
    }
}

Here, the UserService provides a method to retrieve user data based on a given userId. It returns a Mono<User> representing user information.

3.3. Setting up EmailService

Next, let’s add the EmailService:

public class EmailService {
    private final UserService userService;

    public EmailService(UserService userService) {
        this.userService = userService;
    }

    public Mono<Boolean> sendEmail(String userId) {
        return userService.getUser(userId)
          .flatMap(user -> {
              System.out.println("Sending email to: " + user.getEmail());
              return Mono.just(true);
          })
          .defaultIfEmpty(false);
    }
}

As the name suggests, the EmailService is responsible for sending emails to users. Importantly, it depends on the UserService to fetch user details and then send an email based on the retrieved information. The sendEmail() method returns a Mono<Boolean> indicating whether the email was sent successfully.

3.4. Setting up DatabaseService

public class DatabaseService {
    private Map<String, User> dataStore = new ConcurrentHashMap<>();

    public Mono<Boolean> saveUserData(User user) {
        return Mono.create(sink -> {
            try {
                dataStore.put(user.getId(), user);
                sink.success(true);
            } catch (Exception e) {
                sink.success(false);
            }
        });
    }
}

The DatabaseService handles the persistence of user data to a database. We are using a  concurrent map to represent data storage here for simplicity.

It provides a saveUserData() method that takes user information and returns a Mono<Boolean> to signify the success or failure of the database operation.

4. zipWhen() in Action

Now that we have all our services defined, let’s define a controller method that combines the Mono streams from all three services into one response of type Mono<ResponseEntity<String>>. We’ll show how we can use the zipWhen() operator to coordinate various asynchronous operations and convert them all to a single response for the calling client. Let’s define the GET method first:

@GetMapping("/example/{userId}")
public Mono<ResponseEntity<String>> combineAllDataFor(@PathVariable String userId) {
    Mono<User> userMono = userService.getUser(userId);
    Mono<Boolean> emailSentMono = emailService.sendEmail(userId)
      .subscribeOn(Schedulers.parallel());
    Mono<String> databaseResultMono = userMono.flatMap(user -> databaseService.saveUserData(user)
      .map(Object::toString));

    return userMono.zipWhen(user -> emailSentMono, (t1, t2) -> Tuples.of(t1, t2))
      .zipWhen(tuple -> databaseResultMono, (tuple, databaseResult) -> {
          User user = tuple.getT1();
          Boolean emailSent = tuple.getT2();
          return ResponseEntity.ok()
            .body("Response: " + user + ", Email Sent: " + emailSent + ", Database Result: " + databaseResult);
      });
}

When a client calls the GET /example/{userId} endpoint, the userService invokes the combineAllData() method to retrieve information about a user based on the provided userId by calling userService.getUser(userId). This result is stored in Mono<User> called userMono here.

Next, it sends an email to the same user. However, before sending the email, it checks whether the user exists. The result of the email-sending operation (success or failure) is represented by the emailSentMono of type Mono<Boolean>. This operation executes in parallel to save time. It saves the user data (retrieved in step 1) to a database using the databaseService.saveUserData(user). The result of this operation (success or failure) is converted to a string and stored in the Mono<String>.

Importantly, it uses the zipWhen() operator to combine the results from the previous steps. The first zipWhen() combines the user data userMono and the email sending status from emailSentMono into a tuple. The second zipWhen() combines the previous tuple and the database result from dataBaseResultMono to construct a final response. Inside the second zipWhen(), it constructs a response message using the combined data.

The message includes user information, whether the email was successfully sent, and the database operation’s result. In essence, this method orchestrates the retrieval of user data, email sending, and database operations for a specific user and combines the results into a meaningful response, ensuring that everything happens efficiently and concurrently.

5. Testing

Now, let’s put our system under test and verify that the correct response is returned that combines three different types of Reactive Streams:

@Test
public void givenUserId_whenCombineAllData_thenReturnsMonoWithCombinedData() {
    UserService userService = Mockito.mock(UserService.class);
    EmailService emailService = Mockito.mock(EmailService.class);
    DatabaseService databaseService = Mockito.mock(DatabaseService.class);

    String userId = "123";
    User user = new User(userId, "John Doe");

    Mockito.when(userService.getUser(userId))
      .thenReturn(Mono.just(user));
    Mockito.when(emailService.sendEmail(userId))
      .thenReturn(Mono.just(true));
    Mockito.when(databaseService.saveUserData(user))
      .thenReturn(Mono.just(true));

    UserController userController = new UserController(userService, emailService, databaseService);

    Mono<ResponseEntity<String>> responseMono = userController.combineAllDataFor(userId);

    StepVerifier.create(responseMono)
      .expectNextMatches(responseEntity -> responseEntity.getStatusCode() == HttpStatus.OK && responseEntity.getBody()
        .equals("Response: " + user + ", Email Sent: true, Database Result: " + true))
      .verifyComplete();
}