WebFlux: Reactive Programming With Spring, Part 3

The third part of the blog series contains an introduction to WebFlux – Spring's reactive web framework.

Anna Eriksson

Apr. 08, 21 · Tutorial

Likes (18)

Comment

Save

31.6K Views

This is the third part of my blog series on reactive programming, which will give an introduction to WebFlux — Spring’s reactive web framework.

1. An Introduction to Spring Webflux

The original web framework for Spring — Spring Web MVC — was built for the Servlet API and Servlet containers.

WebFlux was introduced as part of Spring Framework 5.0. Unlike Spring MVC, it does not require the Servlet API. It is fully asynchronous and non-blocking, implementing the Reactive Streams specification through the Reactor project (see my previous blog post).

WebFlux requires Reactor as a core dependency but it is also interoperable with other reactive libraries via Reactive Streams.

1.1 Programming Models

Spring WebFlux supports two different programming models: annotation-based and functional.

1.1.1 Annotated Controllers

If you have worked with Spring MVC, the annotation-based model will look quite familiar since it is using the same annotations from the Spring Web module as are being used with Spring MVC. The major difference being that the methods now return the reactive types Mono and Flux. See the following example of a RestController using the annotation-based model:

    Java
   
x

@RestController
@RequestMapping("/students")
public class StudentController {
    @Autowired
    private StudentService studentService;
    public StudentController() {
    }
    @GetMapping("/{id}")
    public Mono<ResponseEntity<Student>> getStudent(@PathVariable long id) {
        return studentService.findStudentById(id)
                .map(ResponseEntity::ok)
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    @GetMapping
    public Flux<Student> listStudents(@RequestParam(name = "name", required = false) String name) {
        return studentService.findStudentsByName(name);
    }
    @PostMapping
    public Mono<Student> addNewStudent(@RequestBody Student student) {
        return studentService.addNewStudent(student);
    }
    @PutMapping("/{id}")
    public Mono<ResponseEntity<Student>> updateStudent(@PathVariable long id, @RequestBody Student student) {
        return studentService.updateStudent(id, student)
                .map(ResponseEntity::ok)
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    @DeleteMapping("/{id}")
    public Mono<ResponseEntity<Void>> deleteStudent(@PathVariable long id) {
        return studentService.findStudentById(id)
                .flatMap(s ->
                        studentService.deleteStudent(s)
                                .then(Mono.just(new ResponseEntity<Void>(HttpStatus.OK)))
                )
                .defaultIfEmpty(new ResponseEntity<>(HttpStatus.NOT_FOUND));
    }
}

Some explanations about the functions used in the example:

The map function is used to transform the item emitted by a Mono by applying a synchronous function to it.
The flatMap function is used to transform the item emitted by the Mono asynchronously, returning the value emitted by another Mono.
The defaultIfEmpty function provides a default value if a Mono is completed without any data.

1.1.2 Functional Endpoints

The functional programming model is lambda-based and leaves the application in charge of the full request handling. It is based on the concepts of HandlerFunctions and RouterFunctions.

HandlerFunctions are used to generate a response for a given request:

    Java
   
xxxxxxxxxx

@FunctionalInterface
public interface HandlerFunction<T extends ServerResponse> {
    Mono<T> handle(ServerRequest request);
}

The RouterFunction is used to route the requests to the HandlerFunctions:

    Java
   
xxxxxxxxxx

@FunctionalInterface
public interface RouterFunction<T extends ServerResponse> {
    Mono<HandlerFunction<T>> route(ServerRequest request);
    ...
}

Continuing with the same student example we would get something like the following using the functional style.

A StudentRouter:

    Java
   
xxxxxxxxxx

@Configuration
public class StudentRouter {
    @Bean
    public RouterFunction<ServerResponse> route(StudentHandler studentHandler){
        return RouterFunctions
            .route(
                GET("/students/{id:[0-9]+}")
                    .and(accept(APPLICATION_JSON)), studentHandler::getStudent)
            .andRoute(
                GET("/students")
                    .and(accept(APPLICATION_JSON)), studentHandler::listStudents)
            .andRoute(
                POST("/students")
                    .and(accept(APPLICATION_JSON)),studentHandler::addNewStudent)
            .andRoute(
                PUT("students/{id:[0-9]+}")
                    .and(accept(APPLICATION_JSON)), studentHandler::updateStudent)
            .andRoute(
                DELETE("/students/{id:[0-9]+}")
                    .and(accept(APPLICATION_JSON)), studentHandler::deleteStudent);
    }
}

And a StudentHandler:

    Java
   
xxxxxxxxxx

@Component
public class StudentHandler {
    private StudentService studentService;
    public StudentHandler(StudentService studentService) {
        this.studentService = studentService;
    }
    public Mono<ServerResponse> getStudent(ServerRequest serverRequest) {
        Mono<Student> studentMono = studentService.findStudentById(
                Long.parseLong(serverRequest.pathVariable("id")));
        return studentMono.flatMap(student -> ServerResponse.ok()
                .body(fromValue(student)))
                .switchIfEmpty(ServerResponse.notFound().build());
    }
    public Mono<ServerResponse> listStudents(ServerRequest serverRequest) {
        String name = serverRequest.queryParam("name").orElse(null);
        return ServerResponse.ok()
                .contentType(MediaType.APPLICATION_JSON)
                .body(studentService.findStudentsByName(name), Student.class);
    }
    public Mono<ServerResponse> addNewStudent(ServerRequest serverRequest) {
        Mono<Student> studentMono = serverRequest.bodyToMono(Student.class);
        return studentMono.flatMap(student ->
                ServerResponse.status(HttpStatus.OK)
                        .contentType(MediaType.APPLICATION_JSON)
                        .body(studentService.addNewStudent(student), Student.class));
    }
    public Mono<ServerResponse> updateStudent(ServerRequest serverRequest) {
        final long studentId = Long.parseLong(serverRequest.pathVariable("id"));
        Mono<Student> studentMono = serverRequest.bodyToMono(Student.class);
        return studentMono.flatMap(student ->
                ServerResponse.status(HttpStatus.OK)
                        .contentType(MediaType.APPLICATION_JSON)
                        .body(studentService.updateStudent(studentId, student), Student.class));
    }
    public Mono<ServerResponse> deleteStudent(ServerRequest serverRequest) {
        final long studentId = Long.parseLong(serverRequest.pathVariable("id"));
        return studentService
                .findStudentById(studentId)
                .flatMap(s -> ServerResponse.noContent().build(studentService.deleteStudent(s)))
                .switchIfEmpty(ServerResponse.notFound().build());
    }
}

Some explanations about the functions used in the example:

The switchIfEmpty function has the same purpose as defaultIfEmpty, but instead of providing a default value, it is used for providing an alternative Mono.

Comparing the two models we can see that:

Using the functional variant requires some more code for things such as retrieving input parameters and parsing to the expected type.
Not relying on annotations, but writing explicit code does offer some more flexibility and could be a better choice if we for example need to implement more complex routing.

1.2 Server Support

WebFlux runs on non-Servlet runtimes such as Netty and Undertow (non-blocking mode) as well as Servlet 3.1+ runtimes such as Tomcat and Jetty.

The Spring Boot WebFlux starter defaults to using Netty, but it is easy to switch by changing your Maven or Gradle dependencies.

For example, to switch to Tomcat, just exclude spring-boot-starter-netty from the spring-boot-starter-webflux dependency and add spring-boot-starter-tomcat:

    XML
   
xxxxxxxxxx

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-netty</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-tomcat</artifactId>
</dependency>

1.3 Configuration

Spring Boot provides auto-configuration for Spring WebFlux that works well for common cases. If you want full control of the WebFlux configuration, the @EnableWebFlux annotation can be used (this annotation would also be needed in a plain Spring application to import the Spring WebFlux configuration).

If you want to keep the Spring Boot WebFlux config and just add some additional WebFlux configuration, you can add your own @Configuration class of type WebFluxConfigurer (but without @EnableWebFlux).

For details and examples, read the WebFlux config documentation.

2. Securing Your Endpoints

To get Spring Security WebFlux support, first, add the spring-boot-starter-security dependency to your project. Now you can enable it by adding the @EnableWebFluxSecurity annotation to your Configuration class (available since Spring Security 5.0).

The following simplified example would add support for two users, one with a USER role and one with an ADMIN role, enforce HTTP basic authentication, and require the ADMIN role for any access to the path /students/admin:

    Java
   
xxxxxxxxxx

@EnableWebFluxSecurity
public class SecurityConfig {
    @Bean
    public MapReactiveUserDetailsService userDetailsService() {
        UserDetails user = User
                .withUsername("user")
                .password(passwordEncoder().encode("userpwd"))
                .roles("USER")
                .build();
        UserDetails admin = User
                .withUsername("admin")
                .password(passwordEncoder().encode("adminpwd"))
                .roles("ADMIN")
                .build();
        return new MapReactiveUserDetailsService(user, admin);
    }
    @Bean
    public SecurityWebFilterChain securityWebFilterChain(ServerHttpSecurity http) {
        return http.authorizeExchange()
                .pathMatchers("/students/admin")
                .hasAuthority("ROLE_ADMIN")
                .anyExchange()
                .authenticated()
                .and().httpBasic()
                .and().build();
    }
    @Bean
    public PasswordEncoder passwordEncoder() {
        return new BCryptPasswordEncoder();
    }
}

It is also possible to secure a method rather than a path, by first adding the annotation @EnableReactiveMethodSecurity to your config:

    Java
   
          x
         
@EnableWebFluxSecurity
@EnableReactiveMethodSecurity
public class SecurityConfig {
    ...
}

And then adding the @PreAuthorize annotation to the methods to be secured. We might for example want our POST, PUT and DELETE methods only to be accessible by the ADMIN role. Then the PreAuthorize annotation could be applied to those methods, like:

    Java
   
xxxxxxxxxx

@DeleteMapping("/{id}")
@PreAuthorize("hasRole('ADMIN')")
public Mono<ResponseEntity<Void>> deleteStudent(@PathVariable long id) {
    ...
}

Spring Security offers more support related to WebFlux applications, such as CSRF protection, OAuth2 integration, and reactive X.509 authentication. For more details, read the following section in the Spring Security documentation: Reactive Applications.

3. WebClient

Spring WebFlux also includes a reactive, fully non-blocking web client. It has a functional, fluent API based on Reactor.

Let’s take a look at a (once again) simplified example, how the WebClient can be used to query our StudentController:

    Java
   
 

     
    

      

     

      

     

     

    
xxxxxxxxxx

          

         

          

         

            
          
public class StudentWebClient {
    WebClient client = WebClient.create("http://localhost:8080");
        public Mono<Student> get(long id) {
            return client
                    .get()
                    .uri("/students/" + id)
                    .headers(headers -> headers.setBasicAuth("user", "userpwd"))
                    .retrieve()
                    .bodyToMono(Student.class);
        }
    
        public Flux<Student> getAll() {
            return client.get()
                    .uri("/students")
                    .headers(headers -> headers.setBasicAuth("user", "userpwd"))
                    .retrieve()
                    .bodyToFlux(Student.class);
        }
    
        public Flux<Student> findByName(String name) {
            return client.get()
                    .uri(uriBuilder -> uriBuilder.path("/students")
                    .queryParam("name", name)
                    .build())
                    .headers(headers -> headers.setBasicAuth("user", "userpwd"))
                    .retrieve()
                    .bodyToFlux(Student.class);
        }
    
        public Mono<Student> create(Student s)  {
            return client.post()
                    .uri("/students")
                    .headers(headers -> headers.setBasicAuth("admin", "adminpwd"))
                    .body(Mono.just(s), Student.class)
                    .retrieve()
                    .bodyToMono(Student.class);
        }
    
        public Mono<Student> update(Student student)  {
            return client
                    .put()
                    .uri("/students/" + student.getId())
                    .headers(headers -> headers.setBasicAuth("admin", "adminpwd"))
                    .body(Mono.just(student), Student.class)
                    .retrieve()
                    .bodyToMono(Student.class);
        }
    
        public Mono<Void> delete(long id) {
            return client
                    .delete()
                    .uri("/students/" + id)
                    .headers(headers -> headers.setBasicAuth("admin", "adminpwd"))
                    .retrieve()
                    .bodyToMono(Void.class);
        }
}

      

     

4. Testing

For testing your reactive web application, WebFlux offers the WebTestClient, which comes with a similar API as the WebClient.

Let’s have a look at how we can test our StudentController using the WebTestClient:

    Java
   
xxxxxxxxxx

@ExtendWith(SpringExtension.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class StudentControllerTest {
    @Autowired
    WebTestClient webClient;
    @Test
    @WithMockUser(roles = "USER")
    void test_getStudents() {
        webClient.get().uri("/students")
                .header(HttpHeaders.ACCEPT, "application/json")
                .exchange()
                .expectStatus().isOk()
                .expectHeader().contentType(MediaType.APPLICATION_JSON)
                .expectBodyList(Student.class);
    }
    @Test
    @WithMockUser(roles = "ADMIN")
    void testAddNewStudent() {
        Student newStudent = new Student();
        newStudent.setName("some name");
        newStudent.setAddress("an address");
        webClient.post().uri("/students")
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.APPLICATION_JSON)
                .body(Mono.just(newStudent), Student.class)
                .exchange()
                .expectStatus().isOk()
                .expectHeader().contentType(MediaType.APPLICATION_JSON)
                .expectBody()
                .jsonPath("$.id").isNotEmpty()
                .jsonPath("$.name").isEqualTo(newStudent.getName())
                .jsonPath("$.address").isEqualTo(newStudent.getAddress());
    }
    ...
}

5. WebSockets and RSocket

5.1 WebSockets

With Spring 5, WebSockets also gets added reactive capabilities. To create a WebSocket server, you can create an implementation of the WebSocketHandler interface, which holds the following method:

    Java
   
xxxxxxxxxx

Mono<Void> handle(WebSocketSession session)

This method is invoked when a new WebSocket connection is established and allows handling of the session. It take a WebSocketSession as input and returns Mono<Void> to signal when application handling of the session is complete.

The WebSocketSession has methods defined for handling the inbound and outbound streams:

    Java
   
xxxxxxxxxx

Flux<WebSocketMessage> receive()
Mono<Void> send(Publisher<WebSocketMessage> messages)

Spring WebFlux also provides a WebSocketClient with implementations for Reactor Netty, Tomcat, Jetty, Undertow, and standard Java.

For more details, read the following chapter in Spring’s Web on Reactive Stack documentation: WebSockets

5.2 RSocket

RSocket is a protocol that models Reactive Streams semantics over a network. It is a binary protocol for use on byte stream transports such as TCP, WebSockets, and Aeron. For an introduction to this topic, I recommend the following blog post that my colleague Pär has written: An introduction to RSocket.

And for more details on Spring Framework’s support for the RSocket protocol: RSocket.

6. To Summarize…

This blog post demonstrated how WebFlux can be used to build a reactive web application. The next and final blog post in this series will show how we can make our entire application stack fully non-blocking by also implementing non-blocking database communication — using R2DBC (Reactive Relational Database Connectivity)!

References

Spring Framework documentation — Web on Reactive Stack

Spring Boot Features — The Spring WebFlux framework

Spring Security — Reactive Applications

Spring Framework Spring Security Reactive programming application Java (programming language) Spring Boot Annotation

Published at DZone with permission of Anna Eriksson. See the original article here.

Opinions expressed by DZone contributors are their own.

Related

Trending