Developing Reactive Applications with Spring Reactive MongoDB

February 21, 2023

We can use MongoDB with Spring Reactive Mongo to develop reactive applications. NoSQL structures seem to be more suitable for the use of reactive programming in production environments compared to R2DBC structures.

Spring Data Mongo Reactive provides us with a reactive MongoDB connector to develop reactive applications. This allows our operations to be performed asynchronously and non-blocking on MongoDB.

In addition to Reactive MongoDB, I recommend reading about the structures that I have previously mentioned and will be using in this article, starting with the theory , reactive endpoints - Mono and Flux , and finally, writing a reactive web client (for testing purposes), and if you are interested, the topic of Reactive Relational Databases.

In this article, we will develop an end-to-end user service using Spring Data Mongo Reactive. We will use Functional Endpoints on Spring WebFlux to prepare our Controller layer, and in our Service layer, we will implement both our own custom queries using ReactiveMongoTemplate and ready-made methods using ReactiveMongoRepository. You can access all the code on the Github repository.

Project Dependencies and Environment Setup

As we always emphasize, if you are developing a reactive application, all its components should be reactive.

implementation 'org.springframework.boot:spring-boot-starter-data-mongodb-reactive'    
implementation 'org.springframework.boot:spring-boot-starter-webflux'    
testImplementation 'org.springframework.boot:spring-boot-starter-test'    
testImplementation 'io.projectreactor:reactor-test'

We can create a docker-compose.yaml to prepare MongoDB, so that we can start up a MongoDB instance that we can use.

version: '3.7'
services:
  mongodb_container:
    image: mongo:latest
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: mongopw
    ports:
      - 27017:27017
    volumes:
      - mongodb_data_container:/data/db

volumes:
  mongodb_data_container:

To make our project work with MongoDB, running the command "docker-compose up" is enough for our MongoDB to be up and running.

Developing Reactive Mongo Application

To enable our Reactive MongoDB to connect to the database we have set up, we need to write the necessary configurations. We can create a Reactive database connection using the MongoConfiguration class that extends the AbstractReactiveMongoConfiguration class. We use the @EnableMongoRepositories annotation to inform Spring that we will be working with MongoDB. In the configuration, we override two methods, one to provide the database information and the other to provide the MongoClient. To create the MongoClient, we use static factory methods and provide the necessary username:password and ip:port information with the .create(connectionString) method.

package dev.gokhana.userservice.configuration;


import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoClients;
import dev.gokhana.userservice.model.User;
import dev.gokhana.userservice.repository.UserRepository;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.config.AbstractReactiveMongoConfiguration;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.data.mongodb.repository.config.EnableMongoRepositories;
import reactor.core.publisher.Flux;

import java.util.concurrent.ThreadLocalRandom;

@Configuration
@EnableMongoRepositories
public class MongoConfiguration extends AbstractReactiveMongoConfiguration {

    @Override
    protected String getDatabaseName() {
        return "users";
    }

    @Override
    @Bean
    public MongoClient reactiveMongoClient() {
        return MongoClients.create("mongodb://root:mongopw@localhost:27017");
    }

    @Bean
    public ReactiveMongoTemplate reactiveMongoTemplate() {
        return new ReactiveMongoTemplate(reactiveMongoClient(), getDatabaseName());
    }

    @Bean
    @ConditionalOnProperty(prefix = "job.autorun", name = "enabled", havingValue = "true", matchIfMissing = true)
    public CommandLineRunner loadData(UserRepository repository) {
        return (args) -> {
            // save a couple of users
            var users = Flux.just(
                    new User("Gökhan", ThreadLocalRandom.current().nextInt(1, 100)),
                    new User("Betül", ThreadLocalRandom.current().nextInt(1, 100)),
                    new User("Zühtü", ThreadLocalRandom.current().nextInt(1, 100))
            );
            repository.saveAll(users).subscribe();
        };
    }
}

We have prepared our structure that will establish a connection to the database. Now it's time to create the object that will be stored in our database. In MongoDB, the data we create is stored in structures called Documents. We can follow the same example we used in my previous article on R2DBC.

@Document
public class User {

    @Id
    private String id;

    @NotBlank
    private String name;

    private Integer score;
   //Getter, Setter, hashCode etc..

Now that we have created the document that will be associated with our database, we can create the repository needed to perform operations on this document. We can create our UserRepository by extending ReactiveMongoRepository, which will provide us with CRUD methods such as findAll, findById, save, and delete.

package dev.gokhana.userservice.repository;

import dev.gokhana.userservice.model.User;
import org.springframework.data.mongodb.repository.ReactiveMongoRepository;

public interface UserRepository extends ReactiveMongoRepository<User, String> {}

In the next step, a service that can manage all operations can be created. UserRepository can be injected into this service. Additionally, ReactiveMongoTemplate can also be made available in the service to demonstrate examples. But first, let's look at how UserRepository is used. We can perform all CRUD operations, and each one contains User object(s) as Mono or Flux.

package dev.gokhana.userservice.service;


import com.mongodb.client.result.DeleteResult;
import dev.gokhana.userservice.model.User;
import dev.gokhana.userservice.repository.UserRepository;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import static org.springframework.data.mongodb.core.query.Criteria.where;
import static org.springframework.data.mongodb.core.query.Query.query;

@Service
public class UserServiceImpl implements UserService {

    private final UserRepository userRepository;
    private final ReactiveMongoTemplate template;

    public UserServiceImpl(UserRepository userRepository, ReactiveMongoTemplate template) {
        this.userRepository = userRepository;
        this.template = template;
    }

    @Override
    public Mono<User> getUserById(String id) {
        return userRepository.findById(id);
    }

    @Override
    public Flux<User> getUsers() {
        return userRepository.findAll();
    }

    @Override
    public Mono<User> saveUser(User userDTO) {
        return userRepository.save(userDTO);
    }

    @Override
    public Mono<User> updateUser(String id, User userDTO) {
        return userRepository.findById(id).flatMap(user -> {
            userDTO.setId(user.getId()); // if there is something else to update do it here.
            return userRepository.save(userDTO);
        });
    }

    @Override
    public Mono<Void> deleteUser(String id) {
        return userRepository.deleteById(id);
    }

    @Override
    public Mono<Long> deleteByName(String name) {
        return template.remove(query(where("name").is(name)), User.class).map(DeleteResult::getDeletedCount);
    }

If we want to manage the entire query infrastructure ourselves, then ReactiveMongoTemplate comes into play. The ReactiveMongoTemplate, which we have already made available for use in our service, provides an infrastructure that allows us to manage all operations ourselves.

@Override
public Mono<Long> deleteByName(String name) {
    return template.remove(query(where("name").is(name)), User.class).map(DeleteResult::getDeletedCount);
}

In the example, we used the remove method with reactiveTemplate. Thus, we have written a method that deletes the names (the name field) that are equal to the parameter received and returns the number of deleted ones. In this way, we can describe the queries we want.

Now, let's prepare functional endpoints that can handle our requests for all service methods. First, we can develop our UserHandler component to include all request and response objects except routing.

package dev.gokhana.userservice.controller;

import dev.gokhana.userservice.controller.validation.ValidatorHandler;
import dev.gokhana.userservice.model.User;
import dev.gokhana.userservice.service.UserService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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 org.springframework.web.util.UriComponentsBuilder;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.util.Map;
import java.util.Optional;

import static org.springframework.web.reactive.function.BodyInserters.fromValue;

@Component
public class UserHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(UserHandler.class);

    private final ValidatorHandler validator;
    private final UserService userService;

    public UserHandler(ValidatorHandler validator, UserService userService) {
        this.validator = validator;
        this.userService = userService;
    }

    public Mono<ServerResponse> getAll(ServerRequest request) {
        return userService.getUsers().collectList().flatMap(users -> {
            if (users.isEmpty()) {
                return ServerResponse.noContent().build();
            }
            return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
                    .body(fromValue(users));
        });
    }

    public Mono<ServerResponse> getUserById(ServerRequest request) {
        String id = request.pathVariable("id");

        return userService.getUserById(id).flatMap(contact -> ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(fromValue(contact))).switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> createUser(ServerRequest request) {
        return request.bodyToMono(User.class)
                .doOnNext(validator::validate)
                .flatMap(userService::saveUser)
                .doOnSuccess(userSaved -> LOGGER.info("User saved with id: {} ", userSaved.getId()))
                .doOnError(e -> LOGGER.error("Error in saveUser method", e))
                .flatMap(user -> ServerResponse.created(getToUri(user)).bodyValue(user));
    }

    public Mono<ServerResponse> updateUser(ServerRequest request) {
        Mono<User> userMono = request.bodyToMono(User.class);
        String id = request.pathVariable("id");
        return userMono
                .flatMap(user -> userService.updateUser(id, user))
                .flatMap(user -> ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
                        .body(fromValue(user))).switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> deleteUser(ServerRequest request) {
        String id = request.pathVariable("id");
        return userService.deleteUser(id)
                .then(ServerResponse.noContent().build())
                .switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> deleteUserByName(ServerRequest request) {
        Optional<String> name = request.queryParam("name");
        if (name.isEmpty())
            return ServerResponse.badRequest().contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(Map.of("error", "There is no name in the field"));
        return userService.deleteByName(name.get())
                .then(ServerResponse.noContent().build())
                .switchIfEmpty(ServerResponse.notFound().build());
    }

    private URI getToUri(User userSaved) {
        return UriComponentsBuilder.fromPath(("/{id}"))
                .buildAndExpand(userSaved.getId()).toUri();
    }
}

As can be seen from the code, our entire structure is ready. Now, we can convert the routing processes into endpoints with RouterFunctions.

package dev.gokhana.userservice.controller;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.MediaType;
import org.springframework.util.StringUtils;
import org.springframework.web.reactive.function.server.RequestPredicates;
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
public class RoutingHandler {

    private static final String API = "/api/v1/users";
    private static final String ID = "/{id}";

    @Bean
    public RouterFunction<ServerResponse> userRouter(UserHandler userHandler) {
        return route(GET(API), userHandler::getAll)
                .andRoute(POST(API).and(accept(MediaType.APPLICATION_JSON)), userHandler::createUser)
                .andRoute(GET(API + ID), userHandler::getUserById)
                .andRoute(PUT(API + ID).and(accept(MediaType.APPLICATION_JSON)), userHandler::updateUser)
                .andRoute(DELETE(API + ID), userHandler::deleteUser)
                .andRoute(DELETE(API).and(RequestPredicates.queryParam("name", StringUtils::hasText)), userHandler::deleteUserByName);
    }

}

Now we have a structure where we can route our requests and we have a reactive application that uses MongoDB end-to-end.

If we want to quickly prepare a unit test structure, we can use WebTestClient and mock our database structures.

package dev.gokhana.userservice;

import dev.gokhana.userservice.model.User;
import dev.gokhana.userservice.repository.UserRepository;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.test.web.reactive.server.WebTestClient;
import reactor.core.publisher.Flux;

import java.util.Arrays;
import java.util.List;

import static org.mockito.Mockito.when;

@SpringBootTest(classes = UserServiceApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, properties = {"job.autorun.enabled=false"})
class UserServiceApplicationTests {

    @Autowired
    private WebTestClient webTestClient;

    @MockBean
    private UserRepository userRepository;

    @Test
    public void findAllTest() {
        List<User> users = Arrays.asList(new User("Gokhan", 15),
                new User("Corç", 25));

        Flux<User> flux = Flux.fromIterable(users);
        when(userRepository.findAll())
                .thenReturn(flux);

        webTestClient.get().uri("api/v1/users")
                .exchange()
                .expectStatus().isOk()
                .expectBodyList(User.class)
                .isEqualTo(users);
    }
}

Now that we have completed all the processes, we can start our service and send requests and examine their results.

POST http://localhost:8083/api/v1/users    
Content-Type: application/json  
 
{    
   "name": "Zühtü",    
   "score": 52    
}

When we send a user creation request, the response we receive will be 201 Created along with the information returned with the object's ID.

{
  "id": "61ed90c1d656605b62e19acd",
  "name": "Zühtü",
  "score": 52
}

Conclusion

With reactive programming, we can perform non-blocking and asynchronous operations with fewer threads and less hardware resources. We added a powerful NoSQL database like MongoDB to this and supported it with a reactive connector to obtain an end-to-end reactive application. In my opinion, applications implemented with MongoDB seem to be more ready for production environments compared to R2DBC database structures.

Developing Reactive Applications with Spring Reactive MongoDB

Project Dependencies and Environment Setup

Developing Reactive Mongo Application

Conclusion

Spring R2DBC for Reactive Relational Databases in Reactive Programming

Spring WebClient: The Next Generation Reactive HTTP Client

Introduction to Reactive Programming with Spring Webflux

What is Reactive Programming? Server-side Reactive Programming

Starting a new journey at Heycar