Build Reactive APIs with Spring WebFlux.

March 21, 2019

Build Reactive APIs with Spring WebFlux.

Spring Boot 2.0 was a long-awaited release from the good folks at Pivotal. One of its new features is reactive web programming support with Spring WebFlux. Spring WebFlux is a web framework that’s built on top of Project Reactor, to give you asynchronous I/O, and allow your application to perform better. If you’re familiar with Spring MVC and building REST APIs, you’ll enjoy Spring WebFlux. There’s just a few basic concepts that are different. Once you know these, you’ll be well on your way to building reactive APIs!

Reactive web programming is great for applications that have streaming data, and clients that consume it and stream it to their users. It ain’t great for developing CRUD apps. If you want to develop a CRUD API, stick with Spring MVC and be happy about it. However, if you’re developing the next Facebook or Twitter with lots of data, a reactive API might be just what you’re looking for.

If you’re not familiar with reactive programming, or just want to see how you can use it with Spring, see the first post in this series: Get Started with Reactive Programming in Spring.

Spring Boot 2.1 is just around the corner, and with it comes Spring Security 5.1 (released this week!). There’s excellent OpenID Connect (OIDC) support in Spring Security 5.1, so we’ll highlight how to use it near the end of this article.

I joined forced with Josh Long to write this post. Josh is a fellow Java Champion, Spring Developer Advocate, and all around fun guy at Pivotal. Josh and I’ve been good friends for a while now, sharing the same passion for Java, developers, and building bleeding-edge applications. We hope you like what we’ve put together for you!

Josh (a.k.a. @starbuxman) authored the Java code in this post, I added the part about securing your API with OIDC. Even though I show how to do it with Okta, it should work with any OIDC provider.

In this post, we’ll show you how to build a REST API using Spring WebFlux. We’ll also show you how you can use WebSockets to provide a stream of real time data from your application.

Get Started with Reactive Programming and Spring WebFlux

Let’s build something! We’ll begin our journey, as usual, at my second favorite place on the internet, the Spring Initializr - start.spring.io. The goal here is to build a new reactive web application that supports reactive data access, and then secure it (reactively!). Select the following dependencies either by using the combo box on the bottom right of the page or by selecting Switch to the Full Version and then choosing DevTools, Reactive Web, Reactive MongoDB. and Lombok.

Figure 1. Selections on the Spring Initializr for a new, reactive application.

This will give you a new project with the following layout.

Example 1. The generated project structure.

.
├── mvnw
├── mvnw.cmd
├── pom.xml
└── src
    ├── main
    │   ├── java
    │   │   └── com
    │   │       └── example
    │   │           └── demo
    │   │               └── DemoApplication.java
    │   └── resources
    │       └── application.properties
    └── test
        └── java
            └── com
                └── example
                    └── demo
                        └── DemoApplicationTests.java

12 directories, 6 files

The Maven build file, pom.xml, is pretty plain, but it assumes we’re going to use JUnit 4. Let’s upgrade JUnit to use JUnit 5, which is a more modern testing framework that’s well supported by Spring Framework 5 and beyond. This owes in no small part to the fact that the most prolific committer of JUnit 5, Sam Brannen, is also the lead of the Spring Test framework.

Add the following dependencies to your new application’s build file, pom.xml: org.junit.jupiter:junit-jupiter-engineand give it a scope of test. Then, exclude the junit:junit dependency from the spring-boot-starter-testdependency. This is the resulting pom.xml:

Example 2. pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>reactive-web</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>demo</name>
    <description>Demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.0.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        <dependency>
            <groupId>de.flapdoodle.embed</groupId>
            <artifactId>de.flapdoodle.embed.mongo</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-engine</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>junit</groupId>
                    <artifactId>junit</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <defaultGoal>spring-boot:run</defaultGoal>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

This is a stock-standard Spring Boot application with a public static void main(String [] args) entry-point class, DemoApplication.java:

Example 3. src/main/java/com/example/demo/DemoApplication.java

package com.example.demo;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
}

This class has a test at src/test/java/com/example/demo/DemoApplicationTests.java that you’ll need to update for JUnit 5.

Example 4. src/test/java/com/example/demo/DemoApplication.java

package com.example.demo;

import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
public class DemoApplicationTests {

    @Test
    public void contextLoads() {
    }
}

There’s also an empty configuration file, src/main/resources/application.properties.

We’re ready to get started! Let’s turn to the first concern, data access.

Add Reactive Data Access with Spring Data

We want to talk to a natively reactive data store. That is, the driver for the database needs to itself support asynchronous I/O, otherwise we won’t be able to scale out reads without scaling out threads, which defeats the point.

Spring Data, an umbrella data access framework, supports a number of reactive data access options including reactive Cassandra, reactive MongoDB, reactive Couchbase and reactive Redis. We’ve chosen MongoDB, so make sure you have a MongoDB database instance running on your local machine on the default host, port, and accessible with the default username and password. If you’re on a Mac, you can use brew install mongodb. If you’re on Debian-based Linux distributions, you can use apt install mongodb.

On a Mac, you’ll need to run the following commands before MongoDB will start.

sudo mkdir -p /data/db
sudo chown -R `id -un` /data/db

MongoDB is a document database, so the unit of interaction is a sparse document - think of it as a JSON stanza that gets persisted and is retrievable by a key (a.k.a., the document ID).

Our application will support manipulating Profile objects. We’re going to persist Profile entities (reactively) using a reactive Spring Data repository, as documents in MongoDB.

If you’d rather see the completed code from this tutorial, you can clone its GitHub repo using the following command:

git clone git@github.com:oktadeveloper/okta-spring-webflux-react-example.git reactive-app

The code in this tutorial is in the reactive-web directory.

To follow along, create a Profile entity class in the com.example.demo package. Give it a single field, email, and another field that will act as the document ID. This entity will be persisted in MongoDB.

Example 5. src/main/java/com/example/demo/Profile.java

package com.example.demo;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;

@Document 
@Data 
@AllArgsConstructor
@NoArgsConstructor
class Profile {

    @Id 
    private String id;

    
    private String email;
}

	`@Document` identifies the entity as a document to be persisted in MongoDB
	`@Data`, `@AllArgsConstructor`, and `@NoArgsConstructor` are all from Lombok. They’re compile-time annotations that tell Lombok to generate getters/setters, constructors, a `toString()` method and an `equals` method.
	`@Id` is a Spring Data annotation that identifies the document ID for this document
	…and finally, this field `email` is the thing that we want to store and retrieve later

In order to persist documents of type Profile, we declaratively define a repository. A repository, a design pattern from Eric Evans' seminal tome, Domain Driven Design, is a way of encapsulating object persistence.

Repositories are responsible for persisting entities and value types. They present clients with a simple model for obtaining persistent objects and managing their life cycle. They decouple application and domain design from persistence technology and strategy choices. They also communicate design decisions about object access. And, finally, they allow easy substitution of implementation with a dummy implementation, ideal in testing. Spring Data’s repositories support all these goals with interface definitions whose implementations are created by the framework at startup time.

Create a Spring Data repository, src/main/java/com/example/demo/ProfileRepository.java.

Example 6. src/main/java/com/example/demo/ProfileRepository.java

package com.example.demo;

import org.springframework.data.mongodb.repository.ReactiveMongoRepository;

interface ProfileRepository extends ReactiveMongoRepository<Profile, String> {
}

This repository extends the Spring Data-provided ReactiveMongoRepository interface which in turn provides a number of data access methods supporting reads, writes, deletes and searches, almost all in terms of method signatures accepting or returning Publisher<T> types.

Example 7. org.springframework.data.mongodb.repository.ReactiveMongoRepository

package org.springframework.data.mongodb.repository;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import org.reactivestreams.Publisher;
import org.springframework.data.domain.Example;
import org.springframework.data.domain.Sort;
import org.springframework.data.repository.NoRepositoryBean;
import org.springframework.data.repository.query.ReactiveQueryByExampleExecutor;
import org.springframework.data.repository.reactive.ReactiveSortingRepository;

@NoRepositoryBean
public interface ReactiveMongoRepository<T, ID> extends ReactiveSortingRepository<T, ID>, ReactiveQueryByExampleExecutor<T> {

    <S extends T> Mono<S> insert(S entity);

    <S extends T> Flux<S> insert(Iterable<S> entities);

    <S extends T> Flux<S> insert(Publisher<S> entities);

    <S extends T> Flux<S> findAll(Example<S> example);

    <S extends T> Flux<S> findAll(Example<S> example, Sort sort);

}

Spring Data will create an object that implements all these methods. It will provide an object for us that we can inject into into other objects to handle persistence. If you define an empty repository, as we have, then there’s little reason to test the repository implementation. It’ll "just work."

Spring Data repositories also supports custom queries. We could, for example, define a custom finder method, of the form Flux<Profile> findByEmail(String email), in our ProfileRepository. This would result in a method being defined that looks for all documents in MongoDB with a predicate that matches the email attribute in the document to the parameter, email, in the method name. If you define custom queries, then this might be an appropriate thing to test.

This is a sample application, of course, so we need some sample data with which to work. Let’s run some initialization logic when the application starts up. We can define a bean of type ApplicationListener<ApplicationReadyEvent> when the application starts us. This will be an enviable opportunity for us to write some sample data into the database once the application’s started up.

Create a SampleDataInitializer.java class to popular the database on startup.

Example 8. src/main/java/com/example/demo/SampleDataInitializer.java

package com.example.demo;

import lombok.extern.log4j.Log4j2;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;

import java.util.UUID;

@Log4j2 
@Component
@org.springframework.context.annotation.Profile("demo") 
class SampleDataInitializer
    implements ApplicationListener<ApplicationReadyEvent> {

    private final ProfileRepository repository; 

    public SampleDataInitializer(ProfileRepository repository) {
        this.repository = repository;
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        repository
            .deleteAll() 
            .thenMany(
                Flux
                    .just("A", "B", "C", "D")
                    .map(name -> new Profile(UUID.randomUUID().toString(), name + "@email.com")) 
                    .flatMap(repository::save) 
            )
            .thenMany(repository.findAll()) 
            .subscribe(profile -> log.info("saving " + profile.toString())); 
    }
}

	a Lombok annotation that results in the creation of a `log` field that is a Log4J logger being added to the class
	this bean initializes sample data that is only useful for a demo. We don’t want this sample data being initialized every time. Spring’s `Profile` annotation tags an object for initialization only when the profile that matches the profile specified in the annotation is specifically activated.
	we’ll use the `ProfileRepository` to handle persistence
	here we start a reactive pipeline by first deleting everything in the database. This operation returns a `Mono<T>`. Both `Mono<T>` and `Flux<T>` support chaining processing with the `thenMany(Publisher<T>)` method. So, after the `deleteAll()` method completes, we then want to process the writes of new data to the database.
	we use Reactor’s `Flux<T>.just(T…)` factory method to create a new `Publisher` with a static list of `String`records, in-memory…
	…and we transform each record in turn into a `Profile` object…
	…that we then persist to the database using our repository
	after all the data has been written to the database, we want to fetch all the records from the database to confirm what we have there
	if we’d stopped at the previous line, the `save` operation, and run this program then we would see… nothing! `Publisher<T>` instances are lazy — you need to `subscribe()` to them to trigger their execution. This last line is where the rubber meets the road. In this case, we’re using the `subscribe(Consumer<T>)` variant that lets us visit every record returned from the `repository.findAll()` operation and print out the record.

You can activate a Spring profile with a command line switch, -Dspring.profiles.active=foo where foo is the name of the profile you’d like to activate. You can also set an environment variable, export SPRING_PROFILES_ACTIVE=foo before running the java process for your Spring Boot application.

You’ll note that in the previous example we use two methods, map(T) and flatMap(T). Map should be familiar if you’ve ever used the Java 8 Stream API. Map visits each record in a publisher and passes it through a lambda function which must transform it. The output of that transformation is then returned and accumulated into a new Publisher. So, the intermediate type after we return from our map operation is a Publisher<Profile>.

In the next line we then call flatMap. flatMap is just like map, except that it unpacks the return value of the lambda given if the value is itself contained in a Publisher<T>. In our case, the repository.save(T) method returns a Mono<T>. If we’d used .map instead of flatMap(T), we’d have a Flux<Mono<T>>, when what we really want is a Flux<T>. We can cleanly solve this problem using flatMap.

Add a Reactive Service

We’re going to use the repository to implement a service that will contain any course grained business logic. In the beginning a lot of the business logic will be pass through logic delegating to the repository, but we can add things like validation and integration with other systems at this layer. Create a ProfileService.java class.

Example 9. src/main/java/com/example/demo/ProfileService.java

package com.example.demo;

import lombok.extern.log4j.Log4j2;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Log4j2
@Service
class ProfileService {

    private final ApplicationEventPublisher publisher; 
    private final ProfileRepository profileRepository; 

    ProfileService(ApplicationEventPublisher publisher, ProfileRepository profileRepository) {
        this.publisher = publisher;
        this.profileRepository = profileRepository;
    }

    public Flux<Profile> all() { 
        return this.profileRepository.findAll();
    }

    public Mono<Profile> get(String id) { 
        return this.profileRepository.findById(id);
    }

    public Mono<Profile> update(String id, String email) { 
        return this.profileRepository
            .findById(id)
            .map(p -> new Profile(p.getId(), email))
            .flatMap(this.profileRepository::save);
    }

    public Mono<Profile> delete(String id) { 
        return this.profileRepository
            .findById(id)
            .flatMap(p -> this.profileRepository.deleteById(p.getId()).thenReturn(p));
    }

    public Mono<Profile> create(String email) { 
        return this.profileRepository
            .save(new Profile(null, email))
            .doOnSuccess(profile -> this.publisher.publishEvent(new ProfileCreatedEvent(profile)));
    }
}

	we’ll want to publish events to other beans managed in the Spring `ApplicationContext`. Earlier, we defined an `ApplicationListener<ApplicationReadyEvent>` that listened for an event that was published in the `ApplicationContext`. Now, we’re going to publish an event for consumption of other beans of our devices in the `ApplicationContext`.
	we defer to our repository to…
	…find all documents or…
	…find a document by its ID…
	…update a `Profile` and give it a new `email`…
	…delete a record by its `id`…
	…or create a new `Profile` in the database and publish an `ApplicationContextEvent`, one of our own creation called `ProfileCreatedEvent`, on successful write to the database. The `doOnSuccess` callback takes a `Consumer<T>` that gets invoked after the data in the reactive pipeline has been written to the database. We’ll see later why this event is so useful.

The ProfileCreatedEvent is just like any other Spring ApplicationEvent.

Example 10. src/main/java/com/example/demo/ProfileCreatedEvent.java

package com.example.demo;

import org.springframework.context.ApplicationEvent;

public class ProfileCreatedEvent extends ApplicationEvent {

    public ProfileCreatedEvent(Profile source) {
        super(source);
    }
}

That wasn’t so bad, was it? Our service was pretty straightforward. The only novelty was the publishing of an event. Everything should be working just fine now. But, of course, we can’t possibly know that unless we test it.

Test Your Reactive Service

Reactive code presents some subtle issues when testing. Remember, our code is asynchronous. It’s possibly concurrent. Each Subscriber<T> could execute on a different thread because the pipeline is managed by a Scheduler. You can change which scheduler is to be used by calling (Flux,Mono).subscribeOn(Scheduler). There’s a convenient factory, Schedulers.\*, that lets you build a new Scheduler from, for example, a java.util.concurrent.Executor. You don’t normally need to override the Scheduler, though. By default there’s one thread per core and the scheduler will just work. You only really need to worry about it when the thing to which you’re subscribing could end up blocking. If, for example, you end up making a call to a blocking JDBC datastore in your Publisher<T>, then you should scale up interactions with that datastore with more threads using a Scheduler.

You need to understand that the Scheduler is present because it implies asynchronicity. This asynchronicity and concurrency is deterministic if you use the operators in the Reactor API: things will execute as they should. It’s only ever problematic, or inscrutable, when attempting to poke at the state of the reactive pipeline from outside.

Then things get a bit twisted. Reactor ships with some very convenient testing support that allow you to assert things about reactive Publisher<T> instances - what is going to be created and when - without having to worry about the schedulers. Let’s look at some tests using the StepVerifier facility.

In order for us to appreciate what’s happening here, we need to take a moment and step back and revisit test slices. Test slices are a feature in Spring Boot that allow the client to load the types in a Spring ApplicationContext that are adjacent to the thing under test.

In this case, we’re interested in testing the data access logic in the service. We are not interested in testing the web functionality. We haven’t even written the web functionality yet, for a start! A test slice lets us tell Spring Boot to load nothing by default and then we can bring pieces back in iteratively.

When Spring Boot starts up it runs a slew of auto-configuration classes. Classes that produce objects that Spring in turn manages for us. The objects are provided by default assuming certain conditions are met. These conditions can include all sorts of things, like the presence of certain types on the classpath, properties in Spring’s Environment, and more. When a Spring Boot application starts up, it is the sum of all the auto-configurations and user configuration given to it. It will be, for our application, database connectivity, object-record mapping (ORM), a webserver, and so much more.

We only need the machinery related to MongoDB and our ProfileService, in isolation. We’ll use the @DataMongoTestannotation to tell Spring Boot to autoconfigure all the things that could be implied in our MongoDB logic, while ignoring things like the web server, runtime and web components.

This results in focused, faster test code that has the benefit of being easier to reproduce. The @DataMongoTest annotation is what’s called a test slice in the Spring Boot world. It supports testing a slice of our application’s functionality in isolation. There are numerous other test slices and you can easily create your own, too.

Test slices can also contribute new auto-configuration supporting tests, specifically. The @DataMongoTest does this. It can even run an embedded MongoDB instance using the Flapdoodle library!

Create ProfileServiceTest to test the logic in your ProfileService.

Example 11. src/test/java/com/example/demo/ProfileServiceTest.java

package com.example.demo;

import lombok.extern.log4j.Log4j2;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.data.mongo.DataMongoTest;
import org.springframework.context.annotation.Import;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

import java.util.UUID;
import java.util.function.Predicate;

@Log4j2
@DataMongoTest 
@Import(ProfileService.class) 
public class ProfileServiceTest {

    private final ProfileService service;
    private final ProfileRepository repository;

    public ProfileServiceTest(@Autowired ProfileService service, 
                              @Autowired ProfileRepository repository) {
        this.service = service;
        this.repository = repository;
    }

    @Test 
    public void getAll() {
        Flux<Profile> saved = repository.saveAll(Flux.just(new Profile(null, "Josh"), new Profile(null, "Matt"), new Profile(null, "Jane")));

        Flux<Profile> composite = service.all().thenMany(saved);

        Predicate<Profile> match = profile -> saved.any(saveItem -> saveItem.equals(profile)).block();

        StepVerifier
            .create(composite) 
            .expectNextMatches(match)  
            .expectNextMatches(match)
            .expectNextMatches(match)
            .verifyComplete(); 
    }

    @Test
    public void save() {
        Mono<Profile> profileMono = this.service.create("email@email.com");
        StepVerifier
            .create(profileMono)
            .expectNextMatches(saved -> StringUtils.hasText(saved.getId()))
            .verifyComplete();
    }

    @Test
    public void delete() {
        String test = "test";
        Mono<Profile> deleted = this.service
            .create(test)
            .flatMap(saved -> this.service.delete(saved.getId()));
        StepVerifier
            .create(deleted)
            .expectNextMatches(profile -> profile.getEmail().equalsIgnoreCase(test))
            .verifyComplete();
    }

    @Test
    public void update() throws Exception {
        Mono<Profile> saved = this.service
            .create("test")
            .flatMap(p -> this.service.update(p.getId(), "test1"));
        StepVerifier
            .create(saved)
            .expectNextMatches(p -> p.getEmail().equalsIgnoreCase("test1"))
            .verifyComplete();
    }

    @Test
    public void getById() {
        String test = UUID.randomUUID().toString();
        Mono<Profile> deleted = this.service
            .create(test)
            .flatMap(saved -> this.service.get(saved.getId()));
        StepVerifier
            .create(deleted)
            .expectNextMatches(profile -> StringUtils.hasText(profile.getId()) && test.equalsIgnoreCase(profile.getEmail()))
            .verifyComplete();
    }
}

	the Spring Boot test slice for MongoDB testing
	we want to add, in addition to all the MongoDB functionality, our custom service for testing
	Look ma! Constructor injection in a unit test!
	Make sure you’re using the new `org.junit.jupiter.api.Test` annotation from JUnit 5.
	In this unit test we setup state in one publisher (`saved`).
	…and then assert things about that state in the various `expectNextMatches` calls
	Make sure to call `verifyComplete`! Otherwise, nothing will happen… and that makes me sad.

We only walked through one test because the rest are unremarkable and similar. You can run mvn test to confirm that the tests work as expected.

The StepVerifier is central to testing all things reactive. It gives us a way to assert that what we think is going to come next in the publisher is in fact going to come next in the publisher. The StepVerifier provides several variants on the expect*theme. Think of this as the reactive equivalent to Assert*.

JUnit 5 supports the same lifecycle methods and annotations (like @Before) as JUnit 4. This is great because it gives you a single place to set up all tests in a class, or to tear down the machinery between tests. That said, I wouldn’t subscribe to any reactive initialization pipelines in the setUp method. Instead, you might define a Flux<T> in the setup, and then compose it in the body of the test methods. This way, you don’t have to wonder if the setup has concluded before the tests themselves execute.

The Web: The Final Frontier

We’ve got a data tier and a service. Let’s stand up RESTful HTTP endpoints to facilitate access to the data. Spring has long had Spring MVC, a web framework that builds upon the Servlet specification. Spring MVC has this concept of a controller - a class that has logic defined in handler methods that process incoming requests and then stages a response - usually a view or a representation of some server-side resource.

In the Spring MVC architecture, requests come in to the web container, they’re routed to the right Servlet (in this case, the Spring MVC DispatcherServlet). The DispatcherServlet then forwards the request to the right handler method in the right controller based on any of a number of configuration details. Those details are typically annotations on the handler methods which themselves live on controller object instances.

Below is the code for a classic Spring MVC style controller that supports manipulating our Profile entities.

Example 12. src/main/java/com/example/demo/GreetingsRestController.java

package com.example.demo;

import org.reactivestreams.Publisher;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;

import java.net.URI;

@RestController 
@RequestMapping(value = "/profiles", produces = MediaType.APPLICATION_JSON_VALUE)  
@org.springframework.context.annotation.Profile("classic")
class ProfileRestController {

    private final MediaType mediaType = MediaType.APPLICATION_JSON_UTF8;
    private final ProfileService profileRepository;

    ProfileRestController(ProfileService profileRepository) {
        this.profileRepository = profileRepository;
    }

    
    @GetMapping
    Publisher<Profile> getAll() {
        return this.profileRepository.all();
    }

    
    @GetMapping("/{id}")
    Publisher<Profile> getById(@PathVariable("id") String id) {
        return this.profileRepository.get(id);
    }

    
    @PostMapping
    Publisher<ResponseEntity<Profile>> create(@RequestBody Profile profile) {
        return this.profileRepository
            .create(profile.getEmail())
            .map(p -> ResponseEntity.created(URI.create("/profiles/" + p.getId()))
                .contentType(mediaType)
                .build());
    }

    @DeleteMapping("/{id}")
    Publisher<Profile> deleteById(@PathVariable String id) {
        return this.profileRepository.delete(id);
    }

    @PutMapping("/{id}")
    Publisher<ResponseEntity<Profile>> updateById(@PathVariable String id, @RequestBody Profile profile) {
        return Mono
            .just(profile)
            .flatMap(p -> this.profileRepository.update(id, p.getEmail()))
            .map(p -> ResponseEntity
                .ok()
                .contentType(this.mediaType)
                .build());
    }
}

	this is yet another stereotype annotation that tells Spring WebFlux that this class provides HTTP handler methods
	There are some attributes that are common to all the HTTP endpoints, like the root URI, and the default `content-type` of all responses produced. You can use `@RequestMapping` to spell this out at the class level and the configuration is inherited for each subordinate handler method
	There are specializations of `@RequestMapping`, one for each HTTP verb, that you can use. This annotation says, "this endpoint is identical to that specified in the root. `@RequestMapping` except that it is limited to HTTP `GET`endpoints"
	This endpoint uses a path variable — a part of the URI that is matched against the incoming request and used to extract a parameter. In this case, it extracts the `id` parameter and makes it available as a method parameter in the handler method.
	This method supports creating a new `Profile` with an HTTP `POST` action. In this handler method we expect incoming requests to have a JSON body that the framework then marshals into a Java object, `Profile`. This happens automatically based on the content-type of the incoming request and the configured, acceptable, convertible payloads supported by Spring WebFlux.

This approach is great if you have a lot of related endpoints that share common dependencies. You can collocate, for example, the GET, PUT, POST, etc., handler logic for a particular resource in one controller class so they can all use the same injected service or validation logic.

The controller approach is not new; Java web frameworks have been using something like it for decades now. The older among us will remember using Apache Struts in the dawn of the 00’s. This approach works well if you have a finite set of HTTP endpoints whose configuration is known a priori. It works well if you want to collocate related endpoints. It also works well if the request matching logic can be described declaratively using Spring’s various annotations.

This approach is also likely to be a perennial favorite for those coming from Spring MVC, as its familiar. Those annotations are exactly the same annotations from Spring MVC. But, this is not Spring MVC. And this isn’t, at least by default, the Servlet API. It’s a brand new web runtime, Spring WebFlux, running - in this instance - on Netty.

Spring Framework 5 changes things, though. Spring Framework 5 assumes a Java 8 baseline and with it lambdas and endless, functional, possibilities!

A lot of what we’re doing in a reactive web application lends itself to the functional programming style. Spring Framework 5 debuts a new functional reactive programming model that mirrors the controller-style programming model in Spring WebFlux. This new programming model is available exclusively in Spring WebFlux. Let’s see an example.

Example 13. src/main/java/com/example/demo/ProfileEndpointConfiguration.java

package com.example.demo;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.server.RequestPredicate;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;

import static org.springframework.web.reactive.function.server.RequestPredicates.*;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;

@Configuration
class ProfileEndpointConfiguration {

    @Bean
    RouterFunction<ServerResponse> routes(ProfileHandler handler) { 
        return route(i(GET("/profiles")), handler::all) 
            .andRoute(i(GET("/profiles/{id}")), handler::getById)
            .andRoute(i(DELETE("/profiles/{id}")), handler::deleteById) 
            .andRoute(i(POST("/profiles")), handler::create)
            .andRoute(i(PUT("/profiles/{id}")), handler::updateById);
    }

    
    private static RequestPredicate i(RequestPredicate target) {
        return new CaseInsensitiveRequestPredicate(target);
    }
}

	This is a Spring bean that describes routes and their handlers to the framework. The handler methods themselves are Java 8 references to methods on another injected bean. They could just as easily have been inline lambdas.
	Each route has a `RequestPredicate` (the object produced by `GET(..)` in this line) and a `HandlerFunction<ServerResponse>`.
	This route uses a path variable, `{id}`, which the framework will use to capture a parameter in the URI string.

We make judicious use of static imports in this example to make things as concise as possible. RouterFunction<ServerResponse> is a builder API. You can store the result of each call to route or andRoute in an intermediate variable if you like. You could loop through records in a for-loop from records in a database and contribute new endpoints dynamically, if you wanted.

Spring WebFlux provides a DSL for describing how to match incoming requests. GET("/profiles") results in a RequestPredicate that matches incoming HTTP GET-method requests that are routed to the URI /profiles. You can compose RequestPredicate instances using .and(RequestPredicate), .not(RequestPredicate), or .or(RequestPredicate). In this example, I also provide a fairly trivial adapter - CaseInsensitiveRequestPredicate - that lower-cases all incoming URLs and matches it against the configured (and lower-cased) URI in the RequestPredicate. The result is that if you type http://localhost:8080/profiles or http://localhost:8080/PROfiLEs they’ll both work.

Example 14. src/main/java/com/example/demo/CaseInsensitiveRequestPredicate.java

package com.example.demo;

import org.springframework.http.server.PathContainer;
import org.springframework.web.reactive.function.server.RequestPredicate;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.support.ServerRequestWrapper;

import java.net.URI;

public class CaseInsensitiveRequestPredicate implements RequestPredicate {

    private final RequestPredicate target;

    CaseInsensitiveRequestPredicate(RequestPredicate target) {
        this.target = target;
    }

    @Override
    public boolean test(ServerRequest request) { 
        return this.target.test(new LowerCaseUriServerRequestWrapper(request));
    }

    @Override
    public String toString() {
        return this.target.toString();
    }
}


class LowerCaseUriServerRequestWrapper extends ServerRequestWrapper {

    LowerCaseUriServerRequestWrapper(ServerRequest delegate) {
        super(delegate);
    }

    @Override
    public URI uri() {
        return URI.create(super.uri().toString().toLowerCase());
    }

    @Override
    public String path() {
        return uri().getRawPath();
    }

    @Override
    public PathContainer pathContainer() {
        return PathContainer.parsePath(path());
    }
}

	The meat of a `RequestPredicate` implementation is in the `test(ServerRequest)` method.
	My implementation wraps the incoming `ServerRequest`, a common enough task that Spring WebFlux even provides a `ServerRequestWrapper`

Once a request is matched, the HandlerFunction<ServerResponse> is invoked to produce a response. Let’s examine our handler object.

Example 15. src/main/java/com/example/demo/ProfileHandler.java

package com.example.demo;

import org.reactivestreams.Publisher;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;

@Component
class ProfileHandler {

    
    private final ProfileService profileService;

    ProfileHandler(ProfileService profileService) {
        this.profileService = profileService;
    }

    
    Mono<ServerResponse> getById(ServerRequest r) {
        return defaultReadResponse(this.profileService.get(id(r)));
    }

    Mono<ServerResponse> all(ServerRequest r) {
        return defaultReadResponse(this.profileService.all());
    }

    Mono<ServerResponse> deleteById(ServerRequest r) {
        return defaultReadResponse(this.profileService.delete(id(r)));
    }

    Mono<ServerResponse> updateById(ServerRequest r) {
        Flux<Profile> id = r.bodyToFlux(Profile.class)
            .flatMap(p -> this.profileService.update(id(r), p.getEmail()));
        return defaultReadResponse(id);
    }

    Mono<ServerResponse> create(ServerRequest request) {
        Flux<Profile> flux = request
            .bodyToFlux(Profile.class)
            .flatMap(toWrite -> this.profileService.create(toWrite.getEmail()));
        return defaultWriteResponse(flux);
    }

    
    private static Mono<ServerResponse> defaultWriteResponse(Publisher<Profile> profiles) {
        return Mono
            .from(profiles)
            .flatMap(p -> ServerResponse
                .created(URI.create("/profiles/" + p.getId()))
                .contentType(MediaType.APPLICATION_JSON_UTF8)
                .build()
            );
    }

    
    private static Mono<ServerResponse> defaultReadResponse(Publisher<Profile> profiles) {
        return ServerResponse
            .ok()
            .contentType(MediaType.APPLICATION_JSON_UTF8)
            .body(profiles, Profile.class);
    }

    private static String id(ServerRequest r) {
        return r.pathVariable("id");
    }
}

	as before, we’re going to make use of our `ProfileService` to do the heavy lifting
	Each handler method has an identical signature: `ServerRequest` is the request parameter and `Mono<ServerResponse>` is the return value.
	We can centralize common logic in, yep! - you guessed it! — functions. This function creates a `Mono<ServerResponse>` from a `Publisher<Profile>` for any incoming request. Each request uses the `ServerResponse` builder object to create a response that has a `Location` header, a `Content-Type` header, and no payload. (You don’t need to send a payload in the response for `PUT` or `POST`, for example).
	this method centralizes all configuration for replies to read requests (for instance, those coming from `GET` verbs)

Straightforward, right? I like this approach - the handler object centralizes processing for related resources into a single class, just like with the controller-style arrangement. We’re also able to centralize routing logic in the @Configuration class.

This means it’s easier to see at a glance what routes have been configured. It’s easier to refactor routing. Routing is also now more dynamic. We can change how requests are matched, and we can dynamically contribute endpoints.

The only drawback to this style is that your code is inextricably tied to the Spring WebFlux component model. Your handler methods in the ProfileHandler are, no question at all, tied to Spring WebFlux. From where I sit, that’s OK.

A controller is supposed to be a thin adapter layer on top of your service. Most of the business logic lives in the service layer, or below. As we’ve already seen, we can easily unit test my service. And anyway, testing my HTTP endpoints requires something

Reference:

Source: https://developer.okta.com/blog/2018/09/24/reactive-apis-with-spring-webflux

Search This Blog

Helmut Steiner