asmhatre/spring-data-demo
1
0
Fork 0
Code Activity

r2dbc: Spring Data R2DBC CRUD, DatabaseClient, reactive transactions and Testcontainers Postgres (post 2)

Browse Source 
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_017ic4b4VQVEJWetTT1dYzyJ
This commit is contained in:
asmhatre
2026-10-29 04:16:34 +00:00
co-authored by Claude Sonnet 5.5
parent 89358e7b7c
commit 7d7d4e872e
27 changed files with 813 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
r2dbc/README.md
+25
View File
@@ -0,0 +1,25 @@
# r2dbc
Companion code for the ankurm.com post
[Reactive Data Access with Spring Data R2DBC: CRUD, Transactions and When to Choose It](https://ankurm.com/spring-data-r2dbc-crud-transactions/).
Spring Boot 4.1.2, JDK 25, Spring Data R2DBC, r2dbc-postgresql 1.1.3.RELEASE, PostgreSQL 17 (Testcontainers 2.0.5).
## What is in here
- `Customer`, `PurchaseOrder`, `Account`: plain records, one row each. No relationship mapping.
- `CustomerRepository`: `ReactiveCrudRepository`, derived queries, `@Query`, `@Modifying`.
- `CustomerOrderService`: three ways to load customers with orders (1+N, two queries, one JOIN with `DatabaseClient`).
- `AccountService`: `@Transactional` and `TransactionalOperator`, with rollback on error.
- `schema.sql`: run by Spring Boot through the R2DBC connection at startup.
- Tests use `StepVerifier` and write the transcripts in `output/`.
## Run
```bash
cd r2dbc
./scripts/run-all.sh # needs Docker; regenerates output/*.txt
```
`output/07-throughput-indicative.txt` is an indicative JDBC vs R2DBC run on a shared 2-vCPU sandbox.
Treat it as a harness, not a result: run it on your own hardware.
r2dbc/output/01-schema-init.txt
+1
View File
@@ -0,0 +1 @@
tables created by schema.sql: [account, customer, purchase_order]
r2dbc/output/02-crud.txt
+6
View File
@@ -0,0 +1,6 @@
save -> id=1 name=Asha Rao
findById -> Customer[id=1, name=Asha Rao, [email protected], tier=GOLD]
update -> name=Asha Rao-Iyer (same id=1, SQL UPDATE)
count -> 1
deleteById-> count now 0
findById(9999) -> empty Mono (completes with no value, no exception)
r2dbc/output/03-lazy-pipeline.txt
+2
View File
@@ -0,0 +1,2 @@
built save(...) pipeline, never subscribed -> rows in table: 0
after .block() on the same pipeline -> rows in table: 1
r2dbc/output/04-queries.txt
+5
View File
@@ -0,0 +1,5 @@
findByTier(GOLD) -> [Asha Rao, Meera Joshi]
@Query searchByName(%RAO%) -> [Asha Rao]
@Modifying changeTier -> 1 row updated
GOLD customers now -> 3
DatabaseClient totalSpent(ravi) -> 200.00
r2dbc/output/05-no-relations.txt
+6
View File
@@ -0,0 +1,6 @@
findAll() on Customer returns rows only. There is no orders field to be lazy or eager:
Customer[id=1, name=A, [email protected], tier=STANDARD]
1+N queries (4 SQL statements for 3 customers) -> [A:2, B:1, C:0]
two queries (2 SQL statements) -> [A:2, B:1, C:0]
one LEFT JOIN, grouped by hand (1 statement) -> [A:2, B:1, C:0]
all three strategies return identical aggregates: true
r2dbc/output/06-transactions.txt
+6
View File
@@ -0,0 +1,6 @@
NO transaction, failure after debit -> alice=60.00 bob=50.00 (money vanished)
@Transactional, failure after debit -> alice=100.00 bob=50.00 (rolled back)
TransactionalOperator, failure -> alice=100.00 bob=50.00 (rolled back)
TransactionalOperator, success -> alice=60.00 bob=90.00 (committed)
@Transactional, CHECK(balance>=0) violated on debit -> DataIntegrityViolationException; alice=100.00 bob=50.00
transfer to a non-existent account -> no error at all; alice=90.00 (debited, credit hit 0 rows)
r2dbc/output/07-throughput-indicative.txt
+5
View File
@@ -0,0 +1,5 @@
INDICATIVE, shared 2-vCPU sandbox, database in a container on the same host.
4000 x 'SELECT ... WHERE id = ?' per round, 100 in flight, pool size 20, 5 rounds after 1 warm-up each.
queries/second, sorted:
R2DBC DatabaseClient : [233, 288, 361, 427, 435] median=361
JDBC + virtual threads : [1240, 1281, 1348, 1558, 3084] median=1348
r2dbc/output/08-missing-testcontainers-r2dbc.txt
+4
View File
@@ -0,0 +1,4 @@
mvn test with testcontainers-r2dbc removed from the pom:
Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 57.47 s <<< FAILURE! -- in com.ankurm.r2dbc.R2dbcDemoTest
Failed to instantiate [io.r2dbc.pool.ConnectionPool]: Factory method 'connectionFactory' threw exception with message: org/testcontainers/r2dbc/R2DBCDatabaseContainer
Caused by: java.lang.ClassNotFoundException: org.testcontainers.r2dbc.R2DBCDatabaseContainer
r2dbc/pom.xml
+97
View File
@@ -0,0 +1,97 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>4.1.2</version>
</parent>
<groupId>com.ankurm</groupId>
<artifactId>r2dbc-demo</artifactId>
<version>1.0.0</version>
<packaging>jar</packaging>
<properties>
<java.version>25</java.version>
<maven.compiler.release>25</maven.compiler.release>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>r2dbc-postgresql</artifactId>
<scope>runtime</scope>
</dependency>
<!-- Test scope only: the JDBC side of the indicative comparison -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-testcontainers</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-postgresql</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-r2dbc</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-bom</artifactId>
<version>2.0.5</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
r2dbc/scripts/repro-missing-artifact.sh
Executable
+20
View File
@@ -0,0 +1,20 @@
#!/usr/bin/env bash
# Shows the real failure when testcontainers-r2dbc is NOT on the test classpath.
# Builds a temporary pom without that dependency, runs one test, keeps the key lines.
set -uo pipefail
cd "$(dirname "$0")/.."
python3 - <<'PY'
import re
s = open('pom.xml').read()
s = re.sub(r'\s*<dependency>\s*<groupId>org.testcontainers</groupId>\s*<artifactId>testcontainers-r2dbc</artifactId>\s*<scope>test</scope>\s*</dependency>', '', s)
open('pom-no-tc-r2dbc.xml', 'w').write(s)
PY
mvn -B -q -T1 -f pom-no-tc-r2dbc.xml test -Dtest='R2dbcDemoTest#schemaCreatedByBootAtStartup' > /tmp/no-tc-r2dbc.log 2>&1
{
echo "mvn test with testcontainers-r2dbc removed from the pom:"
grep -m1 -o 'Tests run:.*' /tmp/no-tc-r2dbc.log
grep -m1 -o "Failed to instantiate \[io.r2dbc.pool.ConnectionPool\]: Factory method 'connectionFactory' threw exception with message: [^ ]*" /tmp/no-tc-r2dbc.log
grep -m1 'Caused by: java.lang.ClassNotFoundException' /tmp/no-tc-r2dbc.log
} > output/08-missing-testcontainers-r2dbc.txt
rm -f pom-no-tc-r2dbc.xml
cat output/08-missing-testcontainers-r2dbc.txt
r2dbc/scripts/run-all.sh
Executable
+9
View File
@@ -0,0 +1,9 @@
#!/usr/bin/env bash
# Regenerates every file in output/ by running the test suite against a real
# PostgreSQL 17 container (Testcontainers). Needs a running Docker daemon.
set -euo pipefail
cd "$(dirname "$0")/.."
rm -f output/*.txt
mvn -B -q test
./scripts/repro-missing-artifact.sh >/dev/null
echo "wrote:"; ls output
r2dbc/src/main/java/com/ankurm/r2dbc/Account.java
+10
View File
@@ -0,0 +1,10 @@
package com.ankurm.r2dbc;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;
import java.math.BigDecimal;
@Table("account")
public record Account(@Id Long id, String owner, BigDecimal balance) {
}
r2dbc/src/main/java/com/ankurm/r2dbc/AccountService.java
+71
View File
@@ -0,0 +1,71 @@
package com.ankurm.r2dbc;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.stereotype.Service;
import org.springframework.transaction.ReactiveTransactionManager;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.reactive.TransactionalOperator;
import reactor.core.publisher.Mono;
import java.math.BigDecimal;
@Service
public class AccountService {
private final DatabaseClient db;
private final TransactionalOperator tx;
public AccountService(DatabaseClient db, ReactiveTransactionManager tm) {
this.db = db;
this.tx = TransactionalOperator.create(tm);
}
private Mono<Long> adjust(long id, BigDecimal delta) {
return db.sql("UPDATE account SET balance = balance + :d WHERE id = :id")
.bind("d", delta).bind("id", id)
.fetch().rowsUpdated();
}
/** Declarative: the returned Mono is wrapped in a transaction by the proxy. */
@Transactional
public Mono<Void> transfer(long from, long to, BigDecimal amount) {
return adjust(from, amount.negate())
.then(adjust(to, amount))
.then();
}
/** Same work, no transaction. The debit commits on its own before the failure. */
public Mono<Void> transferNoTx(long from, long to, BigDecimal amount) {
return adjust(from, amount.negate())
.then(Mono.<Long>error(new IllegalStateException("simulated failure after debit")))
.then(adjust(to, amount))
.then();
}
/** Programmatic: TransactionalOperator wraps exactly the pipeline you hand it. */
public Mono<Void> transferWithOperator(long from, long to, BigDecimal amount, boolean failAfterDebit) {
Mono<Void> work = adjust(from, amount.negate())
.then(failAfterDebit
? Mono.<Long>error(new IllegalStateException("simulated failure after debit"))
: Mono.just(0L))
.then(adjust(to, amount))
.then();
return tx.transactional(work);
}
/** Declarative, failing: the exception rolls the debit back. */
@Transactional
public Mono<Void> transferThenFail(long from, long to, BigDecimal amount) {
return adjust(from, amount.negate())
.then(Mono.<Long>error(new IllegalStateException("simulated failure after debit")))
.then(adjust(to, amount))
.then();
}
public Mono<BigDecimal> balance(long id) {
return db.sql("SELECT balance FROM account WHERE id = :id")
.bind("id", id)
.map(row -> row.get("balance", BigDecimal.class))
.one();
}
}
r2dbc/src/main/java/com/ankurm/r2dbc/Customer.java
+15
View File
@@ -0,0 +1,15 @@
package com.ankurm.r2dbc;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;
/**
* A plain row. Note what is NOT here: no {@code List<PurchaseOrder>} field, no
* @OneToMany. Spring Data R2DBC maps one row to one object and nothing else.
*/
@Table("customer")
public record Customer(@Id Long id, String name, String email, String tier) {
public static Customer ofNew(String name, String email, String tier) {
return new Customer(null, name, email, tier);
}
}
r2dbc/src/main/java/com/ankurm/r2dbc/CustomerOrderService.java
+85
View File
@@ -0,0 +1,85 @@
package com.ankurm.r2dbc;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
* Three ways to get "customers with their orders" when there is no
* relationship mapping. Each makes the cost of that choice visible.
*/
@Service
public class CustomerOrderService {
private final CustomerRepository customers;
private final PurchaseOrderRepository orders;
private final DatabaseClient db;
public CustomerOrderService(CustomerRepository customers, PurchaseOrderRepository orders, DatabaseClient db) {
this.customers = customers;
this.orders = orders;
this.db = db;
}
/** Way 1: 1 + N queries. Correct, readable, and one round trip per customer. */
public Flux<CustomerWithOrders> loadNPlusOne() {
return customers.findAll()
.concatMap(c -> orders.findByCustomerId(c.id()).collectList()
.map(list -> new CustomerWithOrders(c, list)));
}
/** Way 2: two queries total, joined in memory. */
public Flux<CustomerWithOrders> loadTwoQueries() {
return customers.findAll().collectList().flatMapMany(cs ->
orders.findAll().collectList().flatMapMany(os -> {
Map<Long, List<PurchaseOrder>> byCustomer = new LinkedHashMap<>();
os.forEach(o -> byCustomer.computeIfAbsent(o.customerId(), k -> new ArrayList<>()).add(o));
return Flux.fromIterable(cs).map(c ->
new CustomerWithOrders(c, byCustomer.getOrDefault(c.id(), List.of())));
}));
}
/** Way 3: one JOIN with DatabaseClient, grouped by hand. */
public Flux<CustomerWithOrders> loadWithJoin() {
return db.sql("""
SELECT c.id AS cid, c.name, c.email, c.tier,
o.id AS oid, o.total
FROM customer c LEFT JOIN purchase_order o ON o.customer_id = c.id
ORDER BY c.id, o.id
""")
.map((row, meta) -> new Object[]{
new Customer(row.get("cid", Long.class), row.get("name", String.class),
row.get("email", String.class), row.get("tier", String.class)),
row.get("oid", Long.class),
row.get("total", BigDecimal.class)})
.all()
.bufferUntilChanged(r -> ((Customer) r[0]).id())
.map(group -> {
Customer c = (Customer) group.get(0)[0];
List<PurchaseOrder> os = group.stream()
.filter(r -> r[1] != null)
.map(r -> new PurchaseOrder((Long) r[1], c.id(), (BigDecimal) r[2]))
.toList();
return new CustomerWithOrders(c, os);
});
}
/** DatabaseClient with a named parameter and an aggregate - SQL stays SQL. */
public Mono<BigDecimal> totalSpent(String email) {
return db.sql("""
SELECT COALESCE(SUM(o.total), 0) AS spent
FROM customer c JOIN purchase_order o ON o.customer_id = c.id
WHERE c.email = :email
""")
.bind("email", email)
.map(row -> row.get("spent", BigDecimal.class))
.one();
}
}
r2dbc/src/main/java/com/ankurm/r2dbc/CustomerRepository.java
+23
View File
@@ -0,0 +1,23 @@
package com.ankurm.r2dbc;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public interface CustomerRepository extends ReactiveCrudRepository<Customer, Long> {
/** Derived query: Spring Data builds the SQL from the method name. */
Flux<Customer> findByTier(String tier);
Mono<Customer> findByEmail(String email);
/** Hand-written SQL: R2DBC repositories take plain SQL, not JPQL. */
@Query("SELECT * FROM customer WHERE lower(name) LIKE lower(:pattern) ORDER BY name")
Flux<Customer> searchByName(String pattern);
/** Modifying query: must return {@code Mono<Long>} (rows updated), {@code Mono<Boolean>} or {@code Mono<Void>}. */
@org.springframework.data.r2dbc.repository.Modifying
@Query("UPDATE customer SET tier = :tier WHERE id = :id")
Mono<Long> changeTier(Long id, String tier);
}
r2dbc/src/main/java/com/ankurm/r2dbc/CustomerWithOrders.java
+7
View File
@@ -0,0 +1,7 @@
package com.ankurm.r2dbc;
import java.util.List;
/** The aggregate you assemble yourself, because the mapping layer will not. */
public record CustomerWithOrders(Customer customer, List<PurchaseOrder> orders) {
}
r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrder.java
+12
View File
@@ -0,0 +1,12 @@
package com.ankurm.r2dbc;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Column;
import org.springframework.data.relational.core.mapping.Table;
import java.math.BigDecimal;
/** The child row holds only the foreign key value, never a Customer object. */
@Table("purchase_order")
public record PurchaseOrder(@Id Long id, @Column("customer_id") Long customerId, BigDecimal total) {
}
r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrderRepository.java
+8
View File
@@ -0,0 +1,8 @@
package com.ankurm.r2dbc;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import reactor.core.publisher.Flux;
public interface PurchaseOrderRepository extends ReactiveCrudRepository<PurchaseOrder, Long> {
Flux<PurchaseOrder> findByCustomerId(Long customerId);
}
r2dbc/src/main/java/com/ankurm/r2dbc/R2dbcDemoApplication.java
+9
View File
@@ -0,0 +1,9 @@
package com.ankurm.r2dbc;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class R2dbcDemoApplication {
public static void main(String[] args) { SpringApplication.run(R2dbcDemoApplication.class, args); }
}
r2dbc/src/main/resources/application.properties
+5
View File
@@ -0,0 +1,5 @@
# Boot runs schema.sql through the R2DBC ConnectionFactory (not JDBC) at startup.
spring.sql.init.mode=always
spring.r2dbc.pool.initial-size=5
spring.r2dbc.pool.max-size=20
logging.level.org.springframework.r2dbc=INFO
r2dbc/src/main/resources/schema.sql
+22
View File
@@ -0,0 +1,22 @@
DROP TABLE IF EXISTS purchase_order;
DROP TABLE IF EXISTS account;
DROP TABLE IF EXISTS customer;
CREATE TABLE customer (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
email TEXT NOT NULL UNIQUE,
tier TEXT NOT NULL DEFAULT 'STANDARD'
);
CREATE TABLE purchase_order (
id BIGSERIAL PRIMARY KEY,
customer_id BIGINT NOT NULL REFERENCES customer(id),
total NUMERIC(10,2) NOT NULL
);
CREATE TABLE account (
id BIGSERIAL PRIMARY KEY,
owner TEXT NOT NULL,
balance NUMERIC(12,2) NOT NULL CHECK (balance >= 0)
);
r2dbc/src/test/java/com/ankurm/r2dbc/R2dbcDemoTest.java
+218
View File
@@ -0,0 +1,218 @@
package com.ankurm.r2dbc;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.r2dbc.core.DatabaseClient;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.postgresql.PostgreSQLContainer;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import java.math.BigDecimal;
import java.util.List;
import static org.junit.jupiter.api.Assertions.*;
/**
* Every transcript in output/ is written by one of these tests, against a real
* PostgreSQL 17 started by Testcontainers. The schema comes from
* src/main/resources/schema.sql, run by Spring Boot at startup.
*/
@Testcontainers
@SpringBootTest
class R2dbcDemoTest {
@Container
@ServiceConnection
static PostgreSQLContainer postgres = new PostgreSQLContainer("postgres:17-alpine");
@Autowired CustomerRepository customers;
@Autowired PurchaseOrderRepository orders;
@Autowired CustomerOrderService service;
@Autowired AccountService accounts;
@Autowired DatabaseClient db;
@BeforeEach
void clean() {
db.sql("TRUNCATE purchase_order, customer, account RESTART IDENTITY CASCADE").then().block();
}
@Test
void schemaCreatedByBootAtStartup() {
Transcript t = new Transcript("01-schema-init.txt");
List<String> tables = db.sql("""
SELECT table_name FROM information_schema.tables
WHERE table_schema = 'public' ORDER BY table_name""")
.map(r -> r.get("table_name", String.class)).all().collectList().block();
t.line("tables created by schema.sql: %s", tables);
assertEquals(List.of("account", "customer", "purchase_order"), tables);
t.write();
}
@Test
void crudWithStepVerifier() {
Transcript t = new Transcript("02-crud.txt");
StepVerifier.create(customers.save(Customer.ofNew("Asha Rao", "[email protected]", "GOLD")))
.assertNext(c -> { assertNotNull(c.id()); t.line("save -> id=%d name=%s", c.id(), c.name()); })
.verifyComplete();
Customer asha = customers.findByEmail("[email protected]").block();
StepVerifier.create(customers.findById(asha.id()))
.assertNext(c -> t.line("findById -> %s", c))
.verifyComplete();
StepVerifier.create(customers.save(new Customer(asha.id(), "Asha Rao-Iyer", asha.email(), "GOLD")))
.assertNext(c -> t.line("update -> name=%s (same id=%d, SQL UPDATE)", c.name(), c.id()))
.verifyComplete();
StepVerifier.create(customers.count()).expectNext(1L).verifyComplete();
t.line("count -> 1");
StepVerifier.create(customers.deleteById(asha.id())).verifyComplete();
StepVerifier.create(customers.count()).expectNext(0L).verifyComplete();
t.line("deleteById-> count now 0");
StepVerifier.create(customers.findById(9999L)).verifyComplete();
t.line("findById(9999) -> empty Mono (completes with no value, no exception)");
t.write();
}
@Test
void nothingHappensUntilYouSubscribe() {
Transcript t = new Transcript("03-lazy-pipeline.txt");
Mono<Customer> unsubscribed = customers.save(Customer.ofNew("Ghost", "[email protected]", "STANDARD"));
long before = customers.count().block();
t.line("built save(...) pipeline, never subscribed -> rows in table: %d", before);
assertEquals(0L, before);
unsubscribed.block();
long after = customers.count().block();
t.line("after .block() on the same pipeline -> rows in table: %d", after);
assertEquals(1L, after);
t.write();
}
@Test
void derivedAndCustomQueries() {
Transcript t = new Transcript("04-queries.txt");
customers.saveAll(List.of(
Customer.ofNew("Asha Rao", "[email protected]", "GOLD"),
Customer.ofNew("Ravi Kulkarni", "[email protected]", "STANDARD"),
Customer.ofNew("Meera Joshi", "[email protected]", "GOLD"))).blockLast();
StepVerifier.create(customers.findByTier("GOLD").map(Customer::name).collectList())
.assertNext(n -> { t.line("findByTier(GOLD) -> %s", n); assertEquals(2, n.size()); })
.verifyComplete();
StepVerifier.create(customers.searchByName("%RAO%").map(Customer::name))
.expectNext("Asha Rao")
.verifyComplete();
t.line("@Query searchByName(%%RAO%%) -> [Asha Rao]");
Customer ravi = customers.findByEmail("[email protected]").block();
StepVerifier.create(customers.changeTier(ravi.id(), "GOLD")).expectNext(1L).verifyComplete();
t.line("@Modifying changeTier -> 1 row updated");
StepVerifier.create(customers.findByTier("GOLD").count()).expectNext(3L).verifyComplete();
t.line("GOLD customers now -> 3");
orders.saveAll(List.of(
new PurchaseOrder(null, ravi.id(), new BigDecimal("120.50")),
new PurchaseOrder(null, ravi.id(), new BigDecimal("79.50")))).blockLast();
StepVerifier.create(service.totalSpent("[email protected]"))
.assertNext(s -> { t.line("DatabaseClient totalSpent(ravi) -> %s", s); assertEquals(0, s.compareTo(new BigDecimal("200.00"))); })
.verifyComplete();
t.write();
}
@Test
void noRelationsMeansYouAssembleAggregatesYourself() {
Transcript t = new Transcript("05-no-relations.txt");
List<Customer> saved = customers.saveAll(List.of(
Customer.ofNew("A", "[email protected]", "STANDARD"),
Customer.ofNew("B", "[email protected]", "STANDARD"),
Customer.ofNew("C", "[email protected]", "STANDARD"))).collectList().block();
orders.saveAll(List.of(
new PurchaseOrder(null, saved.get(0).id(), new BigDecimal("10.00")),
new PurchaseOrder(null, saved.get(0).id(), new BigDecimal("20.00")),
new PurchaseOrder(null, saved.get(1).id(), new BigDecimal("30.00")))).blockLast();
t.line("findAll() on Customer returns rows only. There is no orders field to be lazy or eager:");
t.line(" %s", customers.findById(saved.get(0).id()).block());
List<String> nPlusOne = shape(service.loadNPlusOne());
List<String> twoQ = shape(service.loadTwoQueries());
List<String> join = shape(service.loadWithJoin());
t.line("1+N queries (4 SQL statements for 3 customers) -> %s", nPlusOne);
t.line("two queries (2 SQL statements) -> %s", twoQ);
t.line("one LEFT JOIN, grouped by hand (1 statement) -> %s", join);
assertEquals(nPlusOne, twoQ);
assertEquals(nPlusOne, join);
t.line("all three strategies return identical aggregates: true");
t.write();
}
private List<String> shape(Flux<CustomerWithOrders> f) {
return f.map(cw -> cw.customer().name() + ":" + cw.orders().size()).collectList().block();
}
@Test
void transactionalRollsBackOnError() {
Transcript t = new Transcript("06-transactions.txt");
seedAccounts();
// 1. no transaction: the debit survives the failure
StepVerifier.create(accounts.transferNoTx(1, 2, new BigDecimal("40.00")))
.expectErrorMessage("simulated failure after debit").verify();
BigDecimal a1 = accounts.balance(1).block(), b1 = accounts.balance(2).block();
t.line("NO transaction, failure after debit -> alice=%s bob=%s (money vanished)", a1, b1);
assertEquals(0, a1.compareTo(new BigDecimal("60.00")));
assertEquals(0, b1.compareTo(new BigDecimal("50.00")));
// 2. @Transactional: rolled back
seedAccounts();
StepVerifier.create(accounts.transferThenFail(1, 2, new BigDecimal("40.00")))
.expectErrorMessage("simulated failure after debit").verify();
BigDecimal a2 = accounts.balance(1).block(), b2 = accounts.balance(2).block();
t.line("@Transactional, failure after debit -> alice=%s bob=%s (rolled back)", a2, b2);
assertEquals(0, a2.compareTo(new BigDecimal("100.00")));
assertEquals(0, b2.compareTo(new BigDecimal("50.00")));
// 3. TransactionalOperator, failure
StepVerifier.create(accounts.transferWithOperator(1, 2, new BigDecimal("40.00"), true))
.expectErrorMessage("simulated failure after debit").verify();
t.line("TransactionalOperator, failure -> alice=%s bob=%s (rolled back)",
accounts.balance(1).block(), accounts.balance(2).block());
assertEquals(0, accounts.balance(1).block().compareTo(new BigDecimal("100.00")));
// 4. success commits
StepVerifier.create(accounts.transferWithOperator(1, 2, new BigDecimal("40.00"), false)).verifyComplete();
t.line("TransactionalOperator, success -> alice=%s bob=%s (committed)",
accounts.balance(1).block(), accounts.balance(2).block());
assertEquals(0, accounts.balance(1).block().compareTo(new BigDecimal("60.00")));
assertEquals(0, accounts.balance(2).block().compareTo(new BigDecimal("90.00")));
// 5. database-raised error (CHECK balance >= 0) on @Transactional
seedAccounts();
StepVerifier.create(accounts.transfer(1, 2, new BigDecimal("500.00")))
.expectError(DataIntegrityViolationException.class).verify();
t.line("@Transactional, CHECK(balance>=0) violated on debit -> DataIntegrityViolationException; alice=%s bob=%s",
accounts.balance(1).block(), accounts.balance(2).block());
// 6. the credit side fails (unknown target is NOT an error: 0 rows updated)
StepVerifier.create(accounts.transfer(1, 999, new BigDecimal("10.00"))).verifyComplete();
t.line("transfer to a non-existent account -> no error at all; alice=%s (debited, credit hit 0 rows)",
accounts.balance(1).block());
t.write();
}
private void seedAccounts() {
db.sql("TRUNCATE account RESTART IDENTITY").then().block();
db.sql("INSERT INTO account(owner, balance) VALUES ('alice', 100.00), ('bob', 50.00)").then().block();
}
}
r2dbc/src/test/java/com/ankurm/r2dbc/ThroughputComparisonTest.java
+112
View File
@@ -0,0 +1,112 @@
package com.ankurm.r2dbc;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.r2dbc.core.DatabaseClient;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.postgresql.PostgreSQLContainer;
import reactor.core.publisher.Flux;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* INDICATIVE ONLY. Same PostgreSQL container, same query, same pool size (20),
* same logical concurrency (100): R2DBC through DatabaseClient versus plain JDBC
* (HikariCP + JdbcTemplate) on virtual threads. Runs on a shared 2-vCPU sandbox
* together with the database itself, so absolute numbers mean little; the point
* is the shape: both are limited by the same 20 connections.
*/
@Testcontainers
@SpringBootTest
class ThroughputComparisonTest {
@Container
@ServiceConnection
static PostgreSQLContainer postgres = new PostgreSQLContainer("postgres:17-alpine");
@Autowired DatabaseClient db;
static final int ROWS = 1000, QUERIES = 4_000, CONCURRENCY = 100, POOL = 20, ROUNDS = 5;
@Test
void compare() throws Exception {
db.sql("INSERT INTO customer(name,email) SELECT 'n'||g, 'n'||g||'@x.io' FROM generate_series(1,%d) g".formatted(ROWS))
.then().block();
HikariConfig cfg = new HikariConfig();
cfg.setJdbcUrl(postgres.getJdbcUrl());
cfg.setUsername(postgres.getUsername());
cfg.setPassword(postgres.getPassword());
cfg.setMaximumPoolSize(POOL);
cfg.setMinimumIdle(POOL);
List<Double> r2 = new ArrayList<>(), jd = new ArrayList<>();
try (HikariDataSource ds = new HikariDataSource(cfg)) {
JdbcTemplate jdbc = new JdbcTemplate(ds);
runR2dbc(); // warm-up, discarded
runJdbc(jdbc); // warm-up, discarded
for (int i = 0; i < ROUNDS; i++) {
r2.add(runR2dbc());
jd.add(runJdbc(jdbc));
}
}
Collections.sort(r2);
Collections.sort(jd);
Transcript t = new Transcript("07-throughput-indicative.txt");
t.line("INDICATIVE, shared 2-vCPU sandbox, database in a container on the same host.");
t.line("%d x 'SELECT ... WHERE id = ?' per round, %d in flight, pool size %d, %d rounds after 1 warm-up each.",
QUERIES, CONCURRENCY, POOL, ROUNDS);
t.line("queries/second, sorted:");
t.line(" R2DBC DatabaseClient : %s median=%.0f", fmt(r2), r2.get(ROUNDS / 2));
t.line(" JDBC + virtual threads : %s median=%.0f", fmt(jd), jd.get(ROUNDS / 2));
t.write();
}
private static String fmt(List<Double> xs) {
return xs.stream().map(d -> String.format("%.0f", d)).toList().toString();
}
private double runR2dbc() {
long t0 = System.nanoTime();
Long n = Flux.range(0, QUERIES)
.flatMap(i -> db.sql("SELECT name FROM customer WHERE id = :id")
.bind("id", 1L + (i % ROWS))
.map(r -> r.get("name", String.class)).one(), CONCURRENCY)
.count().block();
assertEquals(QUERIES, n);
return QUERIES / ((System.nanoTime() - t0) / 1e9);
}
private double runJdbc(JdbcTemplate jdbc) throws Exception {
long t0 = System.nanoTime();
java.util.concurrent.Semaphore inFlight = new java.util.concurrent.Semaphore(CONCURRENCY);
List<Future<String>> fs = new ArrayList<>(QUERIES);
try (ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor()) {
for (int i = 0; i < QUERIES; i++) {
long id = 1L + (i % ROWS);
inFlight.acquire();
fs.add(ex.submit(() -> {
try {
return jdbc.queryForObject("SELECT name FROM customer WHERE id = ?", String.class, id);
} finally { inFlight.release(); }
}));
}
for (Future<String> f : fs) f.get();
}
return QUERIES / ((System.nanoTime() - t0) / 1e9);
}
}
r2dbc/src/test/java/com/ankurm/r2dbc/Transcript.java
+29
View File
@@ -0,0 +1,29 @@
package com.ankurm.r2dbc;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
/** Writes output/NN-*.txt while a test asserts the same numbers. */
final class Transcript {
private final List<String> lines = new ArrayList<>();
private final Path file;
Transcript(String name) { this.file = Path.of("output/" + name); }
Transcript line(String fmt, Object... args) {
String s = fmt.formatted(args);
lines.add(s);
System.out.println(s);
return this;
}
void write() {
try {
Files.createDirectories(file.getParent());
Files.writeString(file, String.join("\n", lines) + "\n");
} catch (IOException e) { throw new RuntimeException(e); }
}
}