Spring Reactive Transaction With Mongo

Introduction

In this blog post we will use ReactiveMongoTransactionManager class to manage the transaction. Along with that we shall use the mongo db.

Version

• Kotlin: 1.3.71
• Spring-boot: 2.2.6.RELEASEA
• Java: 11
• mongo db: 4.2.0-rc2-bionic

Dependencies

You can use Spring Initializer to initialize the project you need following dependencies -

• Spring Reactive Web
• Spring Data Reactive MongoDB

Create docker-compose.yml file

I use docker for running the mongo db the docker compose file looks like this -

version: '3.1'
services:
  mongo:
    container_name: spring-reactive
    image: mongo:4.2.0-rc2-bionic
    ports:
      - 27017:27017
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: example 

Configure MongoDb to the application

Add the following configurartion in the application.yml file.

spring:
  data:
    mongodb:
      uri: mongodb://root:example@localhost:27017
      database: customer
      authentication-database: admin

Create a User document

Create a User class that would have an age field. You shall not allow the user to create an account if the age of the user is below 18.

@Document
data class User(
    @Id
    val id: String,
    val age: Int 
)

Create a data repository for User document

interface CustomerRepo : ReactiveCrudRepository<User, String>

Create a service class

Now create a service class that would have a method called saveAll. The saveAll method will accept the list of an Int array, create the user object, validate the age and the save data to the database.

fun saveAll(vararg ages: Int): Flux<User> {
  return Flux.fromIterable(ages.asIterable())
      .map {
        User(id = UUID.randomUUID().toString(),
            age = it)
      }
      .doOnNext { Assert.isTrue(it.age >= 18, "Age should be greater or equal to 18") }
      .flatMap { userRepo.save(it) }
}

Make the service transactional

To add the transactional support to the application, provide two beans transactionManager and transactionOperator to enable the transactions.

Create a transaction manager

Create a ReactiveTransactionManager that will manages the client sessions for the given ReactiveMongoDatabaseFactory

@Bean
fun transactionManager(rdbf: ReactiveMongoDatabaseFactory): ReactiveMongoTransactionManager {
  return ReactiveMongoTransactionManager(rdbf)
}

Create a transaction operator

Operator class that simplifies programmatic transaction demarcation and transaction exception handling. Create a new TransactionalOperator using ReactiveTransactionManager, using a default transaction.

@Bean
fun transactionOperator(rtm: ReactiveTransactionManager): TransactionalOperator {
  return TransactionalOperator.create(rtm)
}

Add transaction to the service

You can add the transaction by wrapping the callback in the execute method of transactionOperator like below -

val data = Flux.fromIterable(ages.asIterable())
  .map {
    User(id = UUID.randomUUID().toString(),
        age = it)
  }
  .doOnNext { Assert.isTrue(it.age >= 18, "Age should be greater or equal to 18") }
  .flatMap { userRepo.save(it) }

return transactionalOperator.execute{ data} 

OR

You can use annotation @Transactional at the method, which you want to make transactional

@Transactional
fun saveAll(vararg ages: Int): Flux<User> {

Source Code Full source is available at GitHub