Spring Reactive Programming in Java

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 arises: How can we make the Java application in a reactive way. And the answer is using Spring Webflux.

Spring Webflux:

Spring Webflux is a reactive-stack web framework that 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. The reactor is a Reactive Streams library and, therefore, all of its operators support non-blocking back pressure.

Spring Webflux uses two 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 three types of signal translate to calls to a downstream subscriber’s onNext, onComplete, or onError methods.

Flux flux1 = Flux.just(“foo”, “bar”, “foobar”);

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 the subscribe method available, which we need to use as per the need:

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

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

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());
    }
}

Now, run the application:

mvn spring-boot:run

That’s it. I hope this blog helps you better understand reactive programming in Java.

References:

• https://docs.spring.io/spring/docs/current/spring-framework-reference/web-reactive.html#spring-webflux

• https://projectreactor.io/docs/core/release/reference/#core-features

• https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-developing-web-applications.html#boot-features-webflux