Tutorial: Reactive Spring Boot, Part 2: A REST Client for Reactive Streams

This is the second part of our series showing how to build a Reactive application using Spring Boot, Kotlin, Java, and JavaFX. The original inspiration was a 70-minute live demo, which I have split into a series of shorter videos with an accompanying blog post, explaining each of the steps more slowly and in more detail.

This second step shows how to create a Java client that will connect to an endpoint that emits a stream of server-sent events. We'll be using a TDD-inspired process to create the client and test it.

You may also like: A Guide to Streams: In-Depth Tutorial With Examples

This blog post contains a video showing the process step-by-step and a textual walk-through (adapted from the transcript of the video) for those who prefer a written format.

This tutorial is a series of steps during which we will build a full Spring Boot application featuring a Kotlin back-end, a Java client, and a JavaFX user interface.

This second step creates a Reactive Spring Java client that connects to a REST service that's streaming stock prices once a second. This client will be used in later sections of the tutorial.

Create a Project for the Client

We're going to create a new project for this client; we want to keep the client and server code completely separate as they should run completely independently.

1. This project will have multiple modules in, so start by selecting an empty project from the choices on the left of the New Project Wizard.
2. Call the project stock-client and press Finish.
3. By default, IntelliJ IDEA shows the modules section of the Project Structure dialog when a new empty project is created. Add a new module here; this will be a Spring Boot module so choose Spring Initializr on the left.
4. We're using Java 13 as the SDK for this tutorial, although we're not using any of the Java 13 features (you can download JDK 13.0.1 here, then define a new IntelliJ IDEA SDK for it).
5. Enter the group name for the project, and we'll use stock-client as the name.
6. Enter a useful description for the module so it's clear what the purpose of this code is.
7. Keep the defaults of creating a Maven project with Java as the language.
8. We'll select Java 11 as the Java version as this is the most recent Long Term Support version for Java, but for the purposes of this project, it makes no difference.
9. We can optionally change the default package structure if we wish.

Next, we'll select the Spring Boot Starters that we need.

1. Use Spring Boot 2.2.0 RC1 because we'll need some features from this version in later videos of this tutorial.
2. Select the Spring Reactive Web starter and Lombok too.
3. The defaults for module name and location are fine so we'll keep them as they are.

IntelliJ IDEA will use Spring Initializr to create the project and then import it correctly into the IDE. If given the option, enable auto-import on Maven so when we make changes to the pom.xml file the project dependencies will automatically be refreshed.

Create the Client Class

1. Delete the StockClientApplication that Spring Initializr has created for the project, we don't need this for this module, as this module is going to be a library that other applications use not an application in its own right.

public class WebClientStockClient {

}

 2. Create a Java class WebClientStockClient, this is going to use Spring's WebClient to connect to the stock prices service.

Create the Client's Test

One way to drive out the requirements for our client, and to see if it actually works, is to develop it in a test-driven way.

1. Using Ctrl+Shift+T for Windows or Linux (⇧⌘T on macOS) we can navigate to the test for a class. If we do this from WebClientStockClient, we see we don't have one yet for this class. Choose the "Create New Test" option, which will show the Create Test dialog.
2. Choose JUnit5 as the testing framework (note that IntelliJ IDEA offers a range of testing frameworks to choose from).
3. This is actually going to be an end-to-end test so enter WebClientStockClientIntegrationTest as the class name
4. Generate a test method using Alt+Insert (⌘N) and selecting "Test Method" from the generate menu.
5. This is not going to be a perfect example of test-driven development, as we're going to just create a single test that looks at only the best case, sometimes called the happy path. Call the test something like shouldRetrieveStockPricesFromTheService.
6. Create an instance of WebClientStockClient in order to test it.

class WebClientStockClientIntegrationTest {
    @Test
    void shouldRetrieveStockPricesFromTheService() {
        WebClientStockClient webClientStockClient = new WebClientStockClient();
    }
}

One of the things we can do with test-driven development is to code against the API we want, instead of testing something we've already created. IntelliJ IDEA makes this easier because we can create the test to look the way we want, and then generate the correct code from that, usually using Alt+Enter.

1. In the test, call a method pricesFor on WebClientStockClient. This method takes a String that represents the symbol of the stock we want the prices for.

void shouldRetrieveStockPricesFromTheService() {
    WebClientStockClient webClientStockClient = new WebClientStockClient();
    webClientStockClient.pricesFor("SYMBOL");
}

(Note: This code will not compile yet)

Create a Basic Prices Method on the Client

1. (Tip: Press Alt+Enter on the red pricesFor method to get IntelliJ IDEA to create this method on WebClientStockClient, with the expected signature.)
2. Change the method on WebClientStockClient to return a Flux of StockPrice objects.
3. The simplest way to create a method that compiles so we can run our test against it is to get this method to return an empty Flux:

public class WebClientStockClient {
    public Flux<StockPrice> pricesFor(String symbol) {
        return Flux.fromArray(new StockPrice[0]);
    }
}

(Note: This code will not compile yet)

Create a Class to Store Stock Prices

1. (Tip: It's easiest to get IntelliJ IDEA to create the StockPrice class using Alt+Enter on the red StockPrice text.)
2. Create StockPrice in the same package as WebClientStockClient

Here's where we're going to use Lombok. Using Lombok's @Data annotation, we can create a data class similar to our Kotlin data class in the first step of this tutorial. Using this, we only need to define the properties of this class using fields, the getters, setters, equals, hashCode, and toString methods are all provided by Lombok.

Use the Lombok IntelliJ IDEA plugin to get code completion and other useful features when working with Lombok.

1. Add a String symbol, a Double price and a LocalDateTime time to the StockPrice class.
2. Add an @AllArgsConstructor and a @NoArgsConstructor via Lombok, these are needed for our code and for JSON serialization.

@Data
@AllArgsConstructor
@NoArgsConstructor
public class StockPrice {
    private String symbol;
    private Double price;
    private LocalDateTime time;
}

Add Assertions to the Test

We'll go back to WebClientStockClientIntegrationTest and add some assertions, we need to check our returned Flux of prices meets our expectations.

1. Store the returned Flux in a prices local variable.
2. Add an assertion that this is not null.
3. Add an assertion that if we take five prices from the flux, we have more than one price.

@Test
void shouldRetrieveStockPricesFromTheService() {
    // given
    WebClientStockClient webClientStockClient = new WebClientStockClient(webClient);

    // when
    Flux<StockPrice> prices = webClientStockClient.pricesFor("SYMBOL");

    // then
    Assertions.assertNotNull(prices);
    Assertions.assertTrue(prices.take(5).count().block() > 0);
}

When we run this test, we see that it fails. It fails because the Flux has zero elements in it because that's what we hard-coded into the client.

Connect the Client to the Real Service

Let's go back to WebClientStockClient and fill in the implementation.

Now, we want to use WebClient to call our REST service inside our pricesFor method.

1. Remove the stub code from prices for (i.e. delete return Flux.fromArray(new StockPrice[0]);)
2. We're going to use the web client to make a GET request (get()).
3. Give it the URI of our service (http://localhost:8080/stocks/{symbol}) and pass in the symbol.
4. Call retrieve().
5. We need to say how to turn the response of this call into a Flux of some type, so we use bodyToFlux() and give it our data class, StockPrice.class.

public Flux<StockPrice> pricesFor(String symbol) {
    return webClient.get()
                    .uri("http://localhost:8080/stocks/{symbol}", symbol)
                    .retrieve()
                    .bodyToFlux(StockPrice.class);
}

These are the very basic requirements to get a reactive stream from a GET call, but we can also define things like the retry and backoff strategy, remember that understanding the flow of data from publisher to consumer is an important part of creating a successful reactive application.

We can also define what to do when specific Exceptions are thrown. As an example, we can say that when we see an IOException we want to log it. We can use the @Log4j2 annotation from Lombok to give us access to the log and log an error.

This is not the most robust way to handle errors, this simply shows that we can consider Exceptions as a first-class concern in our reactive streams.

import lombok.extern.log4j.Log4j2;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;

import java.io.IOException;
import java.time.Duration;

@Log4j2
public class WebClientStockClient {
    private WebClient webClient;

    public WebClientStockClient(WebClient webClient) {
        this.webClient = webClient;
    }

    public Flux<StockPrice> pricesFor(String symbol) {
        return webClient.get()
                        .uri("http://localhost:8080/stocks/{symbol}", symbol)
                        .retrieve()
                        .bodyToFlux(StockPrice.class)
                        .retryBackoff(5, Duration.ofSeconds(1), Duration.ofSeconds(20))
                        .doOnError(IOException.class, e -> log.error(e.getMessage()));
    }
}

Run the Integration Test

Going back to WebClientStockClientIntegrationTest, we will see there are some things we need to fix.

1. We now need to give our client a WebClient. Create a WebClient field in the test.
2. (Tip: Using Smart Completion, Ctrl+Shift+Space, IntelliJ IDEA will even suggest the full call to the builder that can create a WebClient instance.)

class WebClientStockClientIntegrationTest {
    private WebClient webClient = WebClient.builder().build();

    @Test
    void shouldRetrieveStockPricesFromTheService() {
        WebClientStockClient webClientStockClient = new WebClientStockClient(webClient);

// ...rest of the class

We should see the test go green at this point. If we look at the output, we can see we're decoding StockPrice objects with the symbol that we used in the test, random prices, and a time.

More Thorough Assertions in the Integration Test

This is not the most thorough test, so let's add a bit more detail to our assertions to make sure the client really is doing what we expect. Let's change the assertion to require five prices when we take five prices, and let's make sure that the symbol of one of the stock prices is what we expect.

import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;

class WebClientStockClientIntegrationTest {
    private WebClient webClient = WebClient.builder().build();

    @Test
    void shouldRetrieveStockPricesFromTheService() {
        // given
        WebClientStockClient webClientStockClient = new WebClientStockClient(webClient);

        // when
        Flux<StockPrice> prices = webClientStockClient.pricesFor("SYMBOL");

        // then
        Assertions.assertNotNull(prices);
        Flux<StockPrice> fivePrices = prices.take(5);
        Assertions.assertEquals(5, fivePrices.count().block());
        Assertions.assertEquals("SYMBOL", fivePrices.blockFirst().getSymbol());
    }
}

Summary

Testing reactive applications is a skill in its own right, and there are much better ways to do it than we've shown in this simple example. However, we have successfully used an integration test to drive the API and functionality of our stock prices client.

This client connects to an endpoint that emits server-sent events and returns a Flux of StockPrice objects that could be consumed by another service. We'll show how to do this in later videos in this series. Stay tuned!

The full code is available on GitHub.

Further Reading

A Guide to Streams: In-Depth Tutorial With Examples

Become a Master of Java Streams, Part 1: Creating Streams