R2DBC – Reactive relational database connectivity

R2DBC

R2DBC is a specification designed for reactive programming /stream oriented access to SQL databases. It is fully non-blocking. The main initiative of R2dbc is to establish an open standard on reactive programming with SQL databases. This comprises specifications (R2DBC-SPI) and R2DBC drive implementations.
Here are the current driver implementations supported by R2dbc

1. r2dbc-h2
2. r2dbc-postgres
3. r2dbc-mssql
4. r2dbc-mysql
5. jasync-sql
6. Google cloud-spanner-r2dbc

Why R2DBC
R2DBC has created the need for a non-blocking application to handle concurrency with a small number of threads and scale with fewer hardware resources. It’s an alternate extension for database access instead of standard JDBC API.

Let’s get started
Include the following dependencies to your pom.xml

<dependency>
<groupId>org.springframework.boot.experimental</groupId>
<artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>

<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-r2dbc</artifactId>
</dependency>

R2dbc Configuration
Connection factory – This can be configured via application YAML

spring:
r2dbc:
url: r2dbc:postgres://admin:admin@localhost:5432/customer

ConnectionFactoryInitializer –
This can be initialized from the above configured ConnectionFactory instance. Also, this class can use to populate and initialize the database using SQL scripts. Here is the sample configuration

@Bean
public ConnectionFactoryInitializer initializer(ConnectionFactory connectionFactory){
ConnectionFactoryInitializer initializer = newConnectionFactoryInitializer();
initializer.setConnectionFactory(connectionFactory);

CompositeDatabasePopulator populator = newCompositeDatabasePopulator();
populator.addPopulators(newResourceDatabasePopulator(newClassPathResource(“schema.sql”)));
populator.addPopulators(newResourceDatabasePopulator(newClassPathResource(“test-data.sql”)));
initializer.setDatabasePopulator(populator);
return initializer;
}

ReactiveCrudRepository
This gives full control over the reactive repository and it uses project reactor types and also built on top of Reactive Streams.

public interface CustomerRepository extends ReactiveCrudRepository<Customer,Integer>{

}

Testing with StepVerifier
The StepVerifier will help us understand how data is getting processed in each step of a reactive stream. Verify method will validate all the expectations defined above.

In the below example, verify the expected onNext call count from the stream.

@Test
public void findAll_withExistingRecord_returnAllRecord(){
subject.findAll()
.as(StepVerifier::create)
.expectNextCount(2)
.verifyComplete();
}

How about Transaction
You can always use @Transactional around the method to enable the transaction in a typical annotation way however, R2dbc provide another an option to define this in a programmatic way using TransactionalOperator

serverRequest.bodyToMono(Customer.class)
.flatMap(c->ok().contentType(APPLICATION_JSON)
.body(customerRepository.save(c).as(transactionalOperator::transactional),Customer.class));

You can download the project code from my [GitHub Repository].