Add cqrs module: CQRS in Spring Boot Without a Framework
Separate write/read DataSources, a JPA command side, and two interchangeable projection listeners (sync and @Async) demonstrating the real latency-versus- freshness trade-off CQRS forces. Includes a reflection-based proof that the query side has no dependency on the write side, and a real failure/fix transcript for the -parameters compiler flag this standalone reactor doesn't inherit from spring-boot-starter-parent.
This commit is contained in:
@@ -0,0 +1,26 @@
|
||||
package com.ankurm.cqrsdemo;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.scheduling.annotation.EnableAsync;
|
||||
|
||||
/**
|
||||
* Two stores, one Spring Boot app. The write side is a normal
|
||||
* {@code spring-boot-starter-data-jpa} setup against {@code jdbc:h2:mem:cqrs-write}.
|
||||
* The read side is a second, independent {@link javax.sql.DataSource} against
|
||||
* {@code jdbc:h2:mem:cqrs-read}, reached only through {@link org.springframework.jdbc.core.JdbcTemplate}
|
||||
* — see {@link com.ankurm.cqrsdemo.read.ReadSideConfig}.
|
||||
*
|
||||
* <p>{@code @EnableAsync} backs the {@code async-projection} profile's
|
||||
* {@link com.ankurm.cqrsdemo.read.AsyncOrderSummaryProjection}. Without it the
|
||||
* {@code @Async} annotation on that listener is silently ignored and it runs
|
||||
* synchronously anyway — a real way to lose the "fast write" half of this
|
||||
* trade-off without any error telling you so.
|
||||
*/
|
||||
@SpringBootApplication
|
||||
@EnableAsync
|
||||
public class Application {
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(Application.class, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
package com.ankurm.cqrsdemo.command;
|
||||
|
||||
/** Published after a {@code cancel()} command commits. */
|
||||
public record OrderCancelled(String orderId) {
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
package com.ankurm.cqrsdemo.command;
|
||||
|
||||
import com.ankurm.cqrsdemo.command.internal.OrderLine;
|
||||
import org.springframework.web.bind.annotation.PathVariable;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* The only HTTP entry point into the write side. Note what it returns: just the new
|
||||
* order's id, nothing shaped for display. If a caller wants to show the order back to
|
||||
* someone, that is a query — {@link com.ankurm.cqrsdemo.query.OrderQueryController},
|
||||
* a different controller talking to a different store.
|
||||
*/
|
||||
@RestController
|
||||
public class OrderCommandController {
|
||||
|
||||
private final OrderCommandService commandService;
|
||||
|
||||
public OrderCommandController(OrderCommandService commandService) {
|
||||
this.commandService = commandService;
|
||||
}
|
||||
|
||||
public record LineRequest(String sku, int quantity, long unitPriceCents) {
|
||||
}
|
||||
|
||||
public record PlaceOrderRequest(String customerName, List<LineRequest> lines) {
|
||||
}
|
||||
|
||||
public record PlaceOrderResponse(String orderId) {
|
||||
}
|
||||
|
||||
@PostMapping("/orders")
|
||||
public PlaceOrderResponse placeOrder(@RequestBody PlaceOrderRequest request) {
|
||||
List<OrderLine> lines = request.lines().stream()
|
||||
.map(l -> new OrderLine(l.sku(), l.quantity(), l.unitPriceCents()))
|
||||
.toList();
|
||||
String orderId = commandService.placeOrder(request.customerName(), lines);
|
||||
return new PlaceOrderResponse(orderId);
|
||||
}
|
||||
|
||||
@PostMapping("/orders/{orderId}/ship")
|
||||
public void ship(@PathVariable String orderId) {
|
||||
commandService.shipOrder(orderId);
|
||||
}
|
||||
|
||||
@PostMapping("/orders/{orderId}/cancel")
|
||||
public void cancel(@PathVariable String orderId) {
|
||||
commandService.cancelOrder(orderId);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
package com.ankurm.cqrsdemo.command;
|
||||
|
||||
import com.ankurm.cqrsdemo.command.internal.Order;
|
||||
import com.ankurm.cqrsdemo.command.internal.OrderLine;
|
||||
import com.ankurm.cqrsdemo.command.internal.OrderRepository;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.NoSuchElementException;
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
* Every command method does exactly two things in one transaction: change the
|
||||
* {@code orders}/{@code order_lines} tables, and publish the event that says so.
|
||||
* It never touches a read-model table and has no idea one exists.
|
||||
*
|
||||
* <p>The events are published through plain {@link ApplicationEventPublisher}, not
|
||||
* Spring Modulith's event publication registry that the companion {@code outbox}
|
||||
* module (post <a href="https://ankurm.com/transactional-outbox-spring-modulith-event-publication-registry/">#33</a>)
|
||||
* builds around. That is a deliberate scope line: this module is about the shape of
|
||||
* CQRS itself — two stores, one path in, one path out, a projection connecting them —
|
||||
* not about guaranteeing that projection against a crash between commit and delivery.
|
||||
* Put them together and the registry is what plugs the hole this module's events
|
||||
* leave open; see the "going deeper" link at the end of the first section.
|
||||
*/
|
||||
@Service
|
||||
public class OrderCommandService {
|
||||
|
||||
private final OrderRepository orderRepository;
|
||||
private final ApplicationEventPublisher events;
|
||||
|
||||
public OrderCommandService(OrderRepository orderRepository, ApplicationEventPublisher events) {
|
||||
this.orderRepository = orderRepository;
|
||||
this.events = events;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public String placeOrder(String customerName, List<OrderLine> lines) {
|
||||
String orderId = UUID.randomUUID().toString();
|
||||
Order order = new Order(orderId, customerName, lines);
|
||||
orderRepository.save(order);
|
||||
events.publishEvent(new OrderPlaced(orderId, customerName, lines, order.totalCents()));
|
||||
return orderId;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void shipOrder(String orderId) {
|
||||
Order order = orderRepository.findById(orderId)
|
||||
.orElseThrow(() -> new NoSuchElementException("no such order: " + orderId));
|
||||
order.ship();
|
||||
orderRepository.save(order);
|
||||
events.publishEvent(new OrderShipped(orderId));
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void cancelOrder(String orderId) {
|
||||
Order order = orderRepository.findById(orderId)
|
||||
.orElseThrow(() -> new NoSuchElementException("no such order: " + orderId));
|
||||
order.cancel();
|
||||
orderRepository.save(order);
|
||||
events.publishEvent(new OrderCancelled(orderId));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
package com.ankurm.cqrsdemo.command;
|
||||
|
||||
import com.ankurm.cqrsdemo.command.internal.OrderLine;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Published after the write-side transaction that created the order commits.
|
||||
* This is the only thing the read side ever learns about an order being placed —
|
||||
* it carries everything {@code OrderSummaryProjection} needs so the projection
|
||||
* never has to call back into the write side to fill in a blank.
|
||||
*/
|
||||
public record OrderPlaced(String orderId, String customerName, List<OrderLine> lines, long totalCents) {
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
package com.ankurm.cqrsdemo.command;
|
||||
|
||||
/** Published after a {@code ship()} command commits. */
|
||||
public record OrderShipped(String orderId) {
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package com.ankurm.cqrsdemo.command.internal;
|
||||
|
||||
import jakarta.persistence.CollectionTable;
|
||||
import jakarta.persistence.ElementCollection;
|
||||
import jakarta.persistence.Entity;
|
||||
import jakarta.persistence.EnumType;
|
||||
import jakarta.persistence.Enumerated;
|
||||
import jakarta.persistence.GeneratedValue;
|
||||
import jakarta.persistence.GenerationType;
|
||||
import jakarta.persistence.Id;
|
||||
import jakarta.persistence.JoinColumn;
|
||||
import jakarta.persistence.Table;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* The write-side aggregate. Normalized across two tables ({@code orders} and
|
||||
* {@code orders_lines}), because the write side's only customer is the command
|
||||
* service — nobody queries this shape directly. {@link #status} is an enum stored
|
||||
* as a string so a glance at the {@code orders} table tells you what it means
|
||||
* without a lookup table.
|
||||
*/
|
||||
@Entity
|
||||
@Table(name = "orders")
|
||||
public class Order {
|
||||
|
||||
public enum Status { PLACED, SHIPPED, CANCELLED }
|
||||
|
||||
@Id
|
||||
private String id;
|
||||
|
||||
private String customerName;
|
||||
|
||||
@Enumerated(EnumType.STRING)
|
||||
private Status status;
|
||||
|
||||
@ElementCollection
|
||||
@CollectionTable(name = "order_lines", joinColumns = @JoinColumn(name = "order_id"))
|
||||
private List<OrderLine> lines = new ArrayList<>();
|
||||
|
||||
protected Order() {
|
||||
// JPA
|
||||
}
|
||||
|
||||
public Order(String id, String customerName, List<OrderLine> lines) {
|
||||
this.id = id;
|
||||
this.customerName = customerName;
|
||||
this.lines = new ArrayList<>(lines);
|
||||
this.status = Status.PLACED;
|
||||
}
|
||||
|
||||
public void ship() {
|
||||
this.status = Status.SHIPPED;
|
||||
}
|
||||
|
||||
public void cancel() {
|
||||
this.status = Status.CANCELLED;
|
||||
}
|
||||
|
||||
public String id() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public String customerName() {
|
||||
return customerName;
|
||||
}
|
||||
|
||||
public Status status() {
|
||||
return status;
|
||||
}
|
||||
|
||||
public List<OrderLine> lines() {
|
||||
return Collections.unmodifiableList(lines);
|
||||
}
|
||||
|
||||
public long totalCents() {
|
||||
return lines.stream().mapToLong(OrderLine::lineTotalCents).sum();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package com.ankurm.cqrsdemo.command.internal;
|
||||
|
||||
import jakarta.persistence.Embeddable;
|
||||
|
||||
/**
|
||||
* A single line of a write-side {@link Order}. {@code @Embeddable} rather than its own
|
||||
* entity+table on purpose — the write side's job is to be correct and normalized for
|
||||
* the one thing it does (accept commands), not to be convenient to query. That's the
|
||||
* read side's job.
|
||||
*/
|
||||
@Embeddable
|
||||
public class OrderLine {
|
||||
|
||||
private String sku;
|
||||
private int quantity;
|
||||
private long unitPriceCents;
|
||||
|
||||
protected OrderLine() {
|
||||
// JPA
|
||||
}
|
||||
|
||||
public OrderLine(String sku, int quantity, long unitPriceCents) {
|
||||
this.sku = sku;
|
||||
this.quantity = quantity;
|
||||
this.unitPriceCents = unitPriceCents;
|
||||
}
|
||||
|
||||
public String sku() {
|
||||
return sku;
|
||||
}
|
||||
|
||||
public int quantity() {
|
||||
return quantity;
|
||||
}
|
||||
|
||||
public long unitPriceCents() {
|
||||
return unitPriceCents;
|
||||
}
|
||||
|
||||
public long lineTotalCents() {
|
||||
return (long) quantity * unitPriceCents;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
package com.ankurm.cqrsdemo.command.internal;
|
||||
|
||||
import org.springframework.data.jpa.repository.JpaRepository;
|
||||
|
||||
/**
|
||||
* The write side's only repository. It is package-private in spirit even though Java
|
||||
* cannot enforce that across a single module the way Spring Modulith's {@code internal}
|
||||
* package convention does in the sibling {@code order-fulfillment} module — see
|
||||
* {@link com.ankurm.cqrsdemo.query.OrderQueryController} for the test that checks the
|
||||
* query side never gets a reference to this interface.
|
||||
*/
|
||||
public interface OrderRepository extends JpaRepository<Order, String> {
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package com.ankurm.cqrsdemo.config;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.boot.jdbc.autoconfigure.DataSourceProperties;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
/**
|
||||
* Two independent {@link DataSource} beans from two independent {@code spring.datasource.*}
|
||||
* trees — {@code write} and {@code read} — rather than Boot's single autoconfigured
|
||||
* {@code DataSource}. {@code writeDataSource} is {@code @Primary} so Spring Data JPA's
|
||||
* autoconfiguration (which asks for "the" {@code DataSource} by type) wires to it without
|
||||
* any further configuration; {@code readDataSource} is reachable only by the explicit
|
||||
* {@code @Qualifier("readDataSource")} used in {@link com.ankurm.cqrsdemo.read.ReadSideConfig}
|
||||
* and {@link com.ankurm.cqrsdemo.query.OrderQueryController}. There is no JPA
|
||||
* {@code EntityManagerFactory} pointed at {@code readDataSource} at all — the read side
|
||||
* physically has no Hibernate session to accidentally reuse.
|
||||
*/
|
||||
@Configuration
|
||||
public class DataSourceConfig {
|
||||
|
||||
@Bean
|
||||
@Primary
|
||||
@ConfigurationProperties("spring.datasource.write")
|
||||
public DataSourceProperties writeDataSourceProperties() {
|
||||
return new DataSourceProperties();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@Primary
|
||||
public DataSource writeDataSource(@Qualifier("writeDataSourceProperties") DataSourceProperties properties) {
|
||||
return properties.initializeDataSourceBuilder().build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConfigurationProperties("spring.datasource.read")
|
||||
public DataSourceProperties readDataSourceProperties() {
|
||||
return new DataSourceProperties();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DataSource readDataSource(@Qualifier("readDataSourceProperties") DataSourceProperties properties) {
|
||||
return properties.initializeDataSourceBuilder().build();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package com.ankurm.cqrsdemo.query;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.PathVariable;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Every method here is a {@code SELECT} against {@code order_summary} through the
|
||||
* read-side {@link JdbcTemplate} -- nothing in this class can reach {@code OrderRepository}
|
||||
* or an {@code EntityManager} even by accident, because it is never given one. The
|
||||
* constructor below is the whole proof; the test for it is
|
||||
* {@code OrderQueryControllerHasNoWriteSideAccessTest}, which inspects this exact
|
||||
* constructor through reflection rather than trusting this comment.
|
||||
*/
|
||||
@RestController
|
||||
public class OrderQueryController {
|
||||
|
||||
private final JdbcTemplate readJdbcTemplate;
|
||||
|
||||
public OrderQueryController(@Qualifier("readJdbcTemplate") JdbcTemplate readJdbcTemplate) {
|
||||
this.readJdbcTemplate = readJdbcTemplate;
|
||||
}
|
||||
|
||||
@GetMapping("/order-summaries")
|
||||
public List<OrderSummary> listAll() {
|
||||
return readJdbcTemplate.query("SELECT * FROM order_summary ORDER BY updated_at DESC", this::toSummary);
|
||||
}
|
||||
|
||||
@GetMapping("/order-summaries/{orderId}")
|
||||
public ResponseEntity<OrderSummary> findOne(@PathVariable String orderId) {
|
||||
List<OrderSummary> rows = readJdbcTemplate.query(
|
||||
"SELECT * FROM order_summary WHERE order_id = ?",
|
||||
this::toSummary, orderId);
|
||||
return rows.isEmpty() ? ResponseEntity.notFound().build() : ResponseEntity.ok(rows.get(0));
|
||||
}
|
||||
|
||||
private OrderSummary toSummary(java.sql.ResultSet rs, int rowNum) throws java.sql.SQLException {
|
||||
return new OrderSummary(
|
||||
rs.getString("order_id"),
|
||||
rs.getString("customer_name"),
|
||||
rs.getInt("item_count"),
|
||||
rs.getLong("total_cents"),
|
||||
rs.getString("status"),
|
||||
rs.getTimestamp("updated_at").toInstant());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ankurm.cqrsdemo.query;
|
||||
|
||||
import java.time.Instant;
|
||||
|
||||
/** The read model's own shape -- denormalized, display-ready, nothing like the write side's {@code Order}. */
|
||||
public record OrderSummary(String orderId, String customerName, int itemCount, long totalCents,
|
||||
String status, Instant updatedAt) {
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package com.ankurm.cqrsdemo.read;
|
||||
|
||||
import com.ankurm.cqrsdemo.command.OrderCancelled;
|
||||
import com.ankurm.cqrsdemo.command.OrderPlaced;
|
||||
import com.ankurm.cqrsdemo.command.OrderShipped;
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.event.TransactionPhase;
|
||||
import org.springframework.transaction.event.TransactionalEventListener;
|
||||
|
||||
/**
|
||||
* The {@code async-projection} profile. {@code @Async} on a {@code @TransactionalEventListener}
|
||||
* still waits for the write-side transaction to commit, then hands the actual call off to the
|
||||
* {@code @EnableAsync} executor instead of running it inline -- so the HTTP response for the
|
||||
* command returns as soon as the commit does, without waiting for the read-model write.
|
||||
* The read model is then consistent a little while later, not immediately. See
|
||||
* {@code output/01-async-profile-staleness-window.txt} for a query that lands inside that
|
||||
* window and one that lands after it.
|
||||
*/
|
||||
@Component
|
||||
@Profile("async-projection")
|
||||
class AsyncOrderSummaryProjection {
|
||||
|
||||
private final OrderSummaryWriter writer;
|
||||
|
||||
AsyncOrderSummaryProjection(OrderSummaryWriter writer) {
|
||||
this.writer = writer;
|
||||
}
|
||||
|
||||
@Async
|
||||
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
|
||||
void on(OrderPlaced event) {
|
||||
writer.applyOrderPlaced(event);
|
||||
}
|
||||
|
||||
@Async
|
||||
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
|
||||
void on(OrderShipped event) {
|
||||
writer.applyStatusChange(event.orderId(), "SHIPPED");
|
||||
}
|
||||
|
||||
@Async
|
||||
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
|
||||
void on(OrderCancelled event) {
|
||||
writer.applyStatusChange(event.orderId(), "CANCELLED");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package com.ankurm.cqrsdemo.read;
|
||||
|
||||
import com.ankurm.cqrsdemo.command.OrderPlaced;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.sql.Timestamp;
|
||||
import java.time.Instant;
|
||||
|
||||
/**
|
||||
* The actual read-model write, shared by the synchronous and {@code @Async} projection
|
||||
* listeners so the two profiles differ only in how they are called, never in what gets
|
||||
* written. {@link #simulatedWriteDelayMs} stands in for whatever a real read-model write
|
||||
* costs -- a second database round trip, a Redis call, a search index update -- so the
|
||||
* trade-off the two profiles demonstrate is visible on a stopwatch, not just asserted.
|
||||
*/
|
||||
@Component
|
||||
class OrderSummaryWriter {
|
||||
|
||||
private final JdbcTemplate readJdbcTemplate;
|
||||
private final long simulatedWriteDelayMs;
|
||||
|
||||
OrderSummaryWriter(JdbcTemplate readJdbcTemplate,
|
||||
@Value("${cqrs.projection.simulated-write-delay-ms}") long simulatedWriteDelayMs) {
|
||||
this.readJdbcTemplate = readJdbcTemplate;
|
||||
this.simulatedWriteDelayMs = simulatedWriteDelayMs;
|
||||
}
|
||||
|
||||
void applyOrderPlaced(OrderPlaced event) {
|
||||
sleep();
|
||||
readJdbcTemplate.update("""
|
||||
MERGE INTO order_summary (order_id, customer_name, item_count, total_cents, status, updated_at)
|
||||
KEY (order_id)
|
||||
VALUES (?, ?, ?, ?, 'PLACED', ?)
|
||||
""",
|
||||
event.orderId(), event.customerName(), event.lines().size(), event.totalCents(),
|
||||
Timestamp.from(Instant.now()));
|
||||
}
|
||||
|
||||
void applyStatusChange(String orderId, String status) {
|
||||
sleep();
|
||||
readJdbcTemplate.update(
|
||||
"UPDATE order_summary SET status = ?, updated_at = ? WHERE order_id = ?",
|
||||
status, Timestamp.from(Instant.now()), orderId);
|
||||
}
|
||||
|
||||
private void sleep() {
|
||||
try {
|
||||
Thread.sleep(simulatedWriteDelayMs);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package com.ankurm.cqrsdemo.read;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
/**
|
||||
* Builds the one {@link JdbcTemplate} the read side is allowed to use, against the
|
||||
* {@code readDataSource} bean from {@link com.ankurm.cqrsdemo.config.DataSourceConfig} --
|
||||
* never against the write side's {@code DataSource}, and creates the {@code order_summary}
|
||||
* table it owns on startup. Boot's own {@code JdbcTemplateAutoConfiguration} would hand out
|
||||
* a {@code JdbcTemplate} for the {@code @Primary} (write) {@code DataSource} if anyone asked
|
||||
* for a plain, unqualified one -- which is exactly why
|
||||
* {@link com.ankurm.cqrsdemo.query.OrderQueryController} always asks for this bean by name.
|
||||
*/
|
||||
@Configuration
|
||||
public class ReadSideConfig {
|
||||
|
||||
@Bean
|
||||
public JdbcTemplate readJdbcTemplate(@Qualifier("readDataSource") DataSource readDataSource) {
|
||||
JdbcTemplate jdbcTemplate = new JdbcTemplate(readDataSource);
|
||||
jdbcTemplate.execute("""
|
||||
CREATE TABLE IF NOT EXISTS order_summary (
|
||||
order_id VARCHAR(64) PRIMARY KEY,
|
||||
customer_name VARCHAR(200) NOT NULL,
|
||||
item_count INT NOT NULL,
|
||||
total_cents BIGINT NOT NULL,
|
||||
status VARCHAR(20) NOT NULL,
|
||||
updated_at TIMESTAMP NOT NULL
|
||||
)
|
||||
""");
|
||||
return jdbcTemplate;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package com.ankurm.cqrsdemo.read;
|
||||
|
||||
import com.ankurm.cqrsdemo.command.OrderCancelled;
|
||||
import com.ankurm.cqrsdemo.command.OrderPlaced;
|
||||
import com.ankurm.cqrsdemo.command.OrderShipped;
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.event.TransactionPhase;
|
||||
import org.springframework.transaction.event.TransactionalEventListener;
|
||||
|
||||
/**
|
||||
* The default profile. {@code @TransactionalEventListener} with no {@code @Async} runs on
|
||||
* the same thread that is about to return the HTTP response, after the write-side
|
||||
* transaction has committed. The read model is guaranteed consistent by the time the
|
||||
* command's HTTP response goes out -- at the cost of the command's latency including
|
||||
* the read-model write. See {@code output/00-sync-profile-latency.txt} for what that
|
||||
* actually measures to.
|
||||
*/
|
||||
@Component
|
||||
@Profile("!async-projection")
|
||||
class SyncOrderSummaryProjection {
|
||||
|
||||
private final OrderSummaryWriter writer;
|
||||
|
||||
SyncOrderSummaryProjection(OrderSummaryWriter writer) {
|
||||
this.writer = writer;
|
||||
}
|
||||
|
||||
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
|
||||
void on(OrderPlaced event) {
|
||||
writer.applyOrderPlaced(event);
|
||||
}
|
||||
|
||||
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
|
||||
void on(OrderShipped event) {
|
||||
writer.applyStatusChange(event.orderId(), "SHIPPED");
|
||||
}
|
||||
|
||||
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
|
||||
void on(OrderCancelled event) {
|
||||
writer.applyStatusChange(event.orderId(), "CANCELLED");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
spring.application.name=cqrs
|
||||
|
||||
# The write side: a normal Spring Data JPA setup. @Primary on this bean (see
|
||||
# DataSourceConfig) is what lets JPA autoconfiguration find it with no further wiring.
|
||||
spring.datasource.write.url=jdbc:h2:mem:cqrs-write;DB_CLOSE_DELAY=-1
|
||||
spring.datasource.write.driver-class-name=org.h2.Driver
|
||||
spring.datasource.write.username=sa
|
||||
spring.datasource.write.password=
|
||||
|
||||
spring.jpa.hibernate.ddl-auto=update
|
||||
spring.jpa.open-in-view=false
|
||||
|
||||
# The read side: a second, independent database. Reached only through the
|
||||
# readJdbcTemplate bean in ReadSideConfig -- there is no EntityManagerFactory
|
||||
# pointed at this one.
|
||||
spring.datasource.read.url=jdbc:h2:mem:cqrs-read;DB_CLOSE_DELAY=-1
|
||||
spring.datasource.read.driver-class-name=org.h2.Driver
|
||||
spring.datasource.read.username=sa
|
||||
spring.datasource.read.password=
|
||||
|
||||
# How long the projection pretends the read-model write takes. Large enough to
|
||||
# measure reliably in a test, small enough not to make the test suite slow.
|
||||
cqrs.projection.simulated-write-delay-ms=300
|
||||
@@ -0,0 +1,76 @@
|
||||
package com.ankurm.cqrsdemo;
|
||||
|
||||
import com.ankurm.cqrsdemo.command.OrderCommandController.LineRequest;
|
||||
import com.ankurm.cqrsdemo.command.OrderCommandController.PlaceOrderRequest;
|
||||
import com.ankurm.cqrsdemo.command.OrderCommandController.PlaceOrderResponse;
|
||||
import com.ankurm.cqrsdemo.query.OrderSummary;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.resttestclient.TestRestTemplate;
|
||||
import org.springframework.boot.resttestclient.autoconfigure.AutoConfigureTestRestTemplate;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
import org.springframework.test.context.ActiveProfiles;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
|
||||
/**
|
||||
* {@code async-projection} profile: {@link com.ankurm.cqrsdemo.read.AsyncOrderSummaryProjection}
|
||||
* is active instead. {@code POST /orders} now returns as soon as the write-side transaction
|
||||
* commits, without waiting for the simulated 300ms read-model write -- but that means a query
|
||||
* that lands inside that window sees the order not yet in the read model at all. This test
|
||||
* deliberately queries in both places: immediately (inside the window) and after waiting for
|
||||
* it to close (outside the window), and both results are the real point of the test, not the
|
||||
* final consistent one alone.
|
||||
*/
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
|
||||
@AutoConfigureTestRestTemplate
|
||||
@ActiveProfiles("async-projection")
|
||||
class AsyncProfileStalenessWindowTest {
|
||||
|
||||
@Autowired
|
||||
private TestRestTemplate rest;
|
||||
|
||||
@Test
|
||||
void placingAnOrderReturnsFastAndTheReadModelIsMomentarilyStale() {
|
||||
// Warm-up request, deliberately not measured or asserted on: the very first HTTP
|
||||
// request against a freshly started test context pays for connection-pool startup
|
||||
// and Hibernate/JPA first-use costs that have nothing to do with this profile's
|
||||
// async projection. Measuring the first request directly made the baseline ~370ms
|
||||
// before the projection was even involved -- comfortably past the 300ms this test
|
||||
// is trying to prove the command path *doesn't* pay. Everything after this call
|
||||
// runs against an already-warm context.
|
||||
rest.postForEntity("/orders", new PlaceOrderRequest("Warmup", List.of(new LineRequest("WARMUP", 1, 1))),
|
||||
PlaceOrderResponse.class);
|
||||
|
||||
PlaceOrderRequest request = new PlaceOrderRequest("Dev",
|
||||
List.of(new LineRequest("MECH-KEYBOARD", 1, 8999)));
|
||||
|
||||
long start = System.currentTimeMillis();
|
||||
ResponseEntity<PlaceOrderResponse> placed = rest.postForEntity("/orders", request, PlaceOrderResponse.class);
|
||||
long elapsedMs = System.currentTimeMillis() - start;
|
||||
|
||||
String orderId = placed.getBody().orderId();
|
||||
System.out.println("POST /orders (async profile) took " + elapsedMs + " ms, orderId=" + orderId);
|
||||
assertThat(elapsedMs).isLessThan(300L);
|
||||
|
||||
ResponseEntity<OrderSummary> immediately = rest.getForEntity("/order-summaries/" + orderId, OrderSummary.class);
|
||||
System.out.println("Immediately after POST returned (inside the staleness window), "
|
||||
+ "GET /order-summaries/" + orderId + " -> status=" + immediately.getStatusCode()
|
||||
+ ", body=" + immediately.getBody());
|
||||
assertThat(immediately.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
|
||||
|
||||
await().atMost(Duration.ofSeconds(2)).pollInterval(Duration.ofMillis(50)).untilAsserted(() -> {
|
||||
ResponseEntity<OrderSummary> eventually = rest.getForEntity("/order-summaries/" + orderId, OrderSummary.class);
|
||||
assertThat(eventually.getStatusCode()).isEqualTo(HttpStatus.OK);
|
||||
System.out.println("After waiting for the projection (outside the staleness window), "
|
||||
+ "GET /order-summaries/" + orderId + " -> status=" + eventually.getStatusCode()
|
||||
+ ", body=" + eventually.getBody());
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package com.ankurm.cqrsdemo;
|
||||
|
||||
import com.ankurm.cqrsdemo.query.OrderQueryController;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
|
||||
import java.lang.reflect.Constructor;
|
||||
import java.lang.reflect.Field;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* The query side's version of the hexagonal-architecture post's
|
||||
* {@code mvn dependency:tree}: instead of trusting the Javadoc comment on
|
||||
* {@link OrderQueryController} that says it cannot reach the write side, this inspects
|
||||
* the actual compiled class through reflection. No Spring context needed -- this is a
|
||||
* claim about the class file, not about runtime wiring.
|
||||
*/
|
||||
class QuerySideArchitectureTest {
|
||||
|
||||
@Test
|
||||
void queryControllersOnlyDependencyIsTheReadJdbcTemplate() {
|
||||
Constructor<?>[] constructors = OrderQueryController.class.getDeclaredConstructors();
|
||||
assertThat(constructors).hasSize(1);
|
||||
|
||||
Class<?>[] paramTypes = constructors[0].getParameterTypes();
|
||||
System.out.println("OrderQueryController constructor parameter types: "
|
||||
+ java.util.Arrays.toString(paramTypes));
|
||||
assertThat(paramTypes).containsExactly(JdbcTemplate.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void queryControllerDeclaresNoFieldFromTheWriteSidePackage() {
|
||||
Field[] fields = OrderQueryController.class.getDeclaredFields();
|
||||
for (Field field : fields) {
|
||||
String packageName = field.getType().getPackageName();
|
||||
System.out.println("field " + field.getName() + " : " + field.getType().getName());
|
||||
assertThat(packageName)
|
||||
.as("field %s should not be able to reference the write side", field.getName())
|
||||
.doesNotStartWith("com.ankurm.cqrsdemo.command");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package com.ankurm.cqrsdemo;
|
||||
|
||||
import com.ankurm.cqrsdemo.command.OrderCommandController.LineRequest;
|
||||
import com.ankurm.cqrsdemo.command.OrderCommandController.PlaceOrderRequest;
|
||||
import com.ankurm.cqrsdemo.command.OrderCommandController.PlaceOrderResponse;
|
||||
import com.ankurm.cqrsdemo.query.OrderSummary;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.boot.resttestclient.autoconfigure.AutoConfigureTestRestTemplate;
|
||||
import org.springframework.boot.resttestclient.TestRestTemplate;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* Default profile: no {@code async-projection}, so {@link com.ankurm.cqrsdemo.read.SyncOrderSummaryProjection}
|
||||
* is active. The projection's simulated write delay (300ms, see {@code application.properties})
|
||||
* runs on the same thread that is about to answer the command's HTTP request, so it shows up
|
||||
* directly in how long {@code POST /orders} takes -- and because of that, the read model is
|
||||
* already correct the instant the command returns, with no waiting required.
|
||||
*/
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
|
||||
@AutoConfigureTestRestTemplate
|
||||
class SyncProfileLatencyTest {
|
||||
|
||||
@org.springframework.beans.factory.annotation.Autowired
|
||||
private TestRestTemplate rest;
|
||||
|
||||
@Test
|
||||
void placingAnOrderTakesAtLeastTheSimulatedProjectionDelayAndTheReadModelIsImmediatelyConsistent() {
|
||||
PlaceOrderRequest request = new PlaceOrderRequest("Priya",
|
||||
List.of(new LineRequest("USB-C-CABLE", 2, 1299), new LineRequest("USB-C-HUB", 1, 4999)));
|
||||
|
||||
long start = System.currentTimeMillis();
|
||||
ResponseEntity<PlaceOrderResponse> placed = rest.postForEntity("/orders", request, PlaceOrderResponse.class);
|
||||
long elapsedMs = System.currentTimeMillis() - start;
|
||||
|
||||
String orderId = placed.getBody().orderId();
|
||||
System.out.println("POST /orders (sync profile) took " + elapsedMs + " ms, orderId=" + orderId);
|
||||
assertThat(elapsedMs).isGreaterThanOrEqualTo(300L);
|
||||
|
||||
ResponseEntity<OrderSummary> summary = rest.getForEntity("/order-summaries/" + orderId, OrderSummary.class);
|
||||
System.out.println("Immediately after POST returned, GET /order-summaries/" + orderId
|
||||
+ " -> status=" + summary.getStatusCode() + ", body=" + summary.getBody());
|
||||
assertThat(summary.getBody()).isNotNull();
|
||||
assertThat(summary.getBody().totalCents()).isEqualTo(2 * 1299 + 4999);
|
||||
assertThat(summary.getBody().status()).isEqualTo("PLACED");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user