Reactive Programming : Spring

Table of contents

Reading Time: 3 minutes

Reactive programming is a programming paradigm that promotes an asynchronous, non-blocking, event-driven approach to data processing. Reactive programming involves modeling data and events as observable data streams and implementing data processing routines to react to the changes in those streams.

In the reactive style of programming, we make a request for the resource and start performing other things. When the data is available, we get the notification along with data in the form of call back function. In the callback function, we handle the response as per application/user needs.

Now the question comes, how we can make the Java applications in a reactive way. And the answer is using Spring Webflux.

Spring Webflux:

Spring Webflux is a reactive-stack web framework which is fully non-blocking, supports Reactive Streams back pressure, and runs on such servers as Netty, Undertow, and Servlet 3.1+ containers. It was added in Spring 5.0.

Spring Webflux uses project reactor as a reactive library. Reactor is a Reactive Streams library and, therefore, all of its operators support non-blocking back pressure.

Spring Webflux uses 2 Publishers:

Mono
Flux

Mono:

A Mono is a specialized Publisher that emits at most one item and then optionally terminates with an onComplete signal or an onError signal. In short, it returns 0 or 1 element.

Mono noData = Mono.empty();
Mono data = Mono.just(“rishi”);

Flux:

A Flux is a standard Publisher representing an asynchronous sequence of 0 to N emitted items, optionally terminated by either a completion signal or an error. These 3 types of signal translate to calls to a downstream subscriber’s onNext, onComplete or onError methods.

Flux flux1 = Flux.just(“Ice”, “cream”, “Icecream”);
Flux flux2 = Flux.fromIterable(Arrays.asList(“A”, “B”, “C”));
Flux flux3 = Flux.range(5, 3);
// subscribe
flux.subscribe();

To subscribe, we need to call the subscribe method on Flux. There are different variants of subscribe methods available which we need to use as per the need:

So now, we are familiar with Mono and Flux so let’s proceed with how to create a reactive application with Spring Web Flux.

First of all, we need to add the following in pom.xml:

<parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>2.1.5.RELEASE</version>
</parent>

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
  </dependency>
<dependencies>

Then, we need to define the main class as follows:

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

and then, we need to define the classes for rest APIs. Spring WebFlux comes in two flavors: functional and annotation-based.

Annotation-based:

@RestController
public class UserRestController {

  @Autowired
  private UserRepository userRepository;

  @GetMapping("/users")
    public Flux getUsers() {
        return userRepository.findAll();
    }
    
    @GetMapping("/users/{id}")
    public Mono getUserById(@PathVariable String id) {
        return userRepository.findById(id);
    }
    
    @PostMapping("/users")
    public Mono addUser(@RequestBody User user) {
        return userRepository.save(user);
    }
}

Functional:

In the functional variant, we keep the routing configuration separate from the actual handling of the requests.

We have defined UserRouter for defining routes and UserHandler to handle the request.

UserRouter:

@Configuration
public class UserRouter {
    
    @Bean
    public RouterFunction userRoutes(UserHandler userHandler) {
        
        return RouterFunctions
                .route(RequestPredicates.POST("/users").and(accept(MediaType.APPLICATION_JSON)), userHandler::addUser)
                .andRoute(RequestPredicates.GET("/users").and(accept(MediaType.APPLICATION_JSON)), userHandler::getUsers)
                .andRoute(RequestPredicates.GET("/users/{id}").and(accept(MediaType.APPLICATION_JSON)), userHandler::getUserById);
    }
}

UserHandler:

@Component
public class UserHandler {
    
    @Autowired
    private UserRepository userRepository;
    
    public Mono addUser(ServerRequest request) {
        Mono userMono = request.bodyToMono(User.class);
        return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
                .body(fromPublisher(userMono.flatMap(userRepository::save), User.class));
    }
    
    public Mono getUsers(ServerRequest request) {
        return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
                .body(userRepository.findAll(), User.class)
                .switchIfEmpty(ServerResponse.notFound().build());
    }
public Mono getUserById(ServerRequest request) {
        String id = request.pathVariable("id");
        return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
                .body(userRepository.findById(id), User.class)
                .switchIfEmpty(ServerResponse.notFound().build());
    }
}

Run application: