Add core-events module: @EventListener, @TransactionalEventListener phases and async listeners on Boot 4.1
Thirteen captured transcripts: a listener is a blocking method call, ordering and chaining, SpEL conditions, generic-event erasure, every transaction phase on commit and rollback, what an AFTER_COMMIT listener can write, which exceptions reach the publisher, async listeners on platform and virtual threads, and the annotations behind Spring Modulith's @ApplicationModuleListener. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Uu7q8vPeREyT4218EJPzz1
This commit is contained in:
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.events.asyncdemo;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication(scanBasePackages = {"com.ankurm.events.asyncdemo", "com.ankurm.events.support"})
|
||||
public class AsyncApp {
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
package com.ankurm.events.asyncdemo;
|
||||
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.scheduling.annotation.EnableAsync;
|
||||
|
||||
/** Without this annotation @Async does nothing: see the last section of output/09. */
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableAsync
|
||||
public class AsyncConfig {
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package com.ankurm.events.asyncdemo;
|
||||
|
||||
import static org.springframework.transaction.event.TransactionPhase.AFTER_COMMIT;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import com.ankurm.events.support.Visibility;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.event.TransactionalEventListener;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
|
||||
@Component
|
||||
public class AsyncListeners {
|
||||
|
||||
private final Gate gate;
|
||||
private final Visibility visibility;
|
||||
|
||||
public AsyncListeners(Gate gate, Visibility visibility) {
|
||||
this.gate = gate;
|
||||
this.visibility = visibility;
|
||||
}
|
||||
|
||||
/** Waits until the publisher has moved on, which it can only do if this listener did not block it. */
|
||||
@Async
|
||||
@EventListener
|
||||
void slow(OrderPlaced event) throws InterruptedException {
|
||||
gate.publishReturned.await(5, TimeUnit.SECONDS);
|
||||
Trace.add("@Async @EventListener: running on %s, after publishEvent returned", Trace.where());
|
||||
gate.slowDone.countDown();
|
||||
}
|
||||
|
||||
/** Looks for the publisher's row while the publisher's transaction is still open. */
|
||||
@Async
|
||||
@EventListener
|
||||
void readsTheRow(RowCheck event) {
|
||||
Trace.add("@Async @EventListener: transaction active on this thread: %s; row visible: %s",
|
||||
TransactionSynchronizationManager.isActualTransactionActive(), visibility.orderVisibleElsewhere(event.id()));
|
||||
gate.plainDone.countDown();
|
||||
}
|
||||
|
||||
/** The same look, but not until the publisher has committed. */
|
||||
@Async
|
||||
@TransactionalEventListener(phase = AFTER_COMMIT)
|
||||
void readsTheRowAfterCommit(RowCheck event) {
|
||||
Trace.add("@Async @TransactionalEventListener(AFTER_COMMIT): on %s; row visible: %s", Trace.where(),
|
||||
visibility.orderVisibleElsewhere(event.id()));
|
||||
gate.commitDone.countDown();
|
||||
}
|
||||
|
||||
@Async
|
||||
@EventListener
|
||||
void boom(Exploding event) {
|
||||
throw new IllegalStateException("async listener blew up");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package com.ankurm.events.asyncdemo;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.jdbc.core.simple.JdbcClient;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
@Service
|
||||
public class AsyncOrderService {
|
||||
|
||||
private final JdbcClient jdbc;
|
||||
private final ApplicationEventPublisher publisher;
|
||||
private final Gate gate;
|
||||
|
||||
public AsyncOrderService(JdbcClient jdbc, ApplicationEventPublisher publisher, Gate gate) {
|
||||
this.jdbc = jdbc;
|
||||
this.publisher = publisher;
|
||||
this.gate = gate;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void place(String id) throws InterruptedException {
|
||||
jdbc.sql("insert into orders(id, amount) values (?, 1)").param(id).update();
|
||||
publisher.publishEvent(new RowCheck(id));
|
||||
// Hold the transaction open until the plain async listener has looked, so its answer cannot be a race.
|
||||
gate.plainDone.await(5, TimeUnit.SECONDS);
|
||||
Trace.add("AsyncOrderService.place: leaving the method, commit follows");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.asyncdemo;
|
||||
|
||||
public record Exploding(String id) {
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package com.ankurm.events.asyncdemo;
|
||||
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/** Latches that make the timing in the scenarios deterministic. */
|
||||
@Component
|
||||
public class Gate {
|
||||
|
||||
/** Released by the test once {@code publishEvent} has returned. */
|
||||
public volatile CountDownLatch publishReturned = new CountDownLatch(1);
|
||||
public volatile CountDownLatch slowDone = new CountDownLatch(1);
|
||||
public volatile CountDownLatch plainDone = new CountDownLatch(1);
|
||||
public volatile CountDownLatch commitDone = new CountDownLatch(1);
|
||||
|
||||
public void reset() {
|
||||
publishReturned = new CountDownLatch(1);
|
||||
slowDone = new CountDownLatch(1);
|
||||
plainDone = new CountDownLatch(1);
|
||||
commitDone = new CountDownLatch(1);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.asyncdemo;
|
||||
|
||||
public record OrderPlaced(String id) {
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.asyncdemo;
|
||||
|
||||
public record RowCheck(String id) {
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.events.asyncplain;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class AsyncPlainApp {
|
||||
}
|
||||
+17
@@ -0,0 +1,17 @@
|
||||
package com.ankurm.events.asyncplain;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/** Marked @Async, in an application that never says @EnableAsync. */
|
||||
@Component
|
||||
public class ForgottenEnableAsyncListener {
|
||||
|
||||
@Async
|
||||
@EventListener
|
||||
void listen(OrderPlaced event) {
|
||||
Trace.add("@Async @EventListener: running on %s", Trace.where());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.asyncplain;
|
||||
|
||||
public record OrderPlaced(String id) {
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.events.badcondition;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class BadConditionApp {
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.badcondition;
|
||||
|
||||
public record OrderPlaced(String id, int amount) {
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
package com.ankurm.events.badcondition;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
public class TypoListener {
|
||||
|
||||
/** "amountt" is not a property of OrderPlaced. */
|
||||
@EventListener(condition = "#event.amountt >= 100")
|
||||
void large(OrderPlaced event) {
|
||||
Trace.add("never reached");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package com.ankurm.events.basics;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
public class AuditListener {
|
||||
|
||||
@EventListener
|
||||
void audit(OrderPlaced event) {
|
||||
Trace.add("AuditListener.audit: received %s on %s", event, Trace.where());
|
||||
}
|
||||
|
||||
/** A listener that does not need the event can name the type in the annotation instead. */
|
||||
@EventListener(OrderPlaced.class)
|
||||
void counted() {
|
||||
Trace.add("AuditListener.counted: no parameter, still called");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ankurm.events.basics;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
/** Scans only this package, so each scenario starts with exactly its own beans. */
|
||||
@SpringBootApplication
|
||||
public class BasicsApp {
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
package com.ankurm.events.basics;
|
||||
|
||||
/** An event. A plain record: nothing to extend and nothing to implement. */
|
||||
public record OrderPlaced(String id, int amount) {
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package com.ankurm.events.basics;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
@Service
|
||||
public class OrderService {
|
||||
|
||||
private final ApplicationEventPublisher publisher;
|
||||
|
||||
public OrderService(ApplicationEventPublisher publisher) {
|
||||
this.publisher = publisher;
|
||||
}
|
||||
|
||||
public void place(String id, int amount) {
|
||||
Trace.add("OrderService.place: publishing on %s", Trace.where());
|
||||
publisher.publishEvent(new OrderPlaced(id, amount));
|
||||
Trace.add("OrderService.place: publishEvent returned");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.events.conditional;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class ConditionalApp {
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package com.ankurm.events.conditional;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
public class ConditionalListeners {
|
||||
|
||||
@EventListener(condition = "#event.amount >= 100")
|
||||
@Order(1)
|
||||
void large(OrderPlaced event) {
|
||||
Trace.add("large (amount >= 100)");
|
||||
}
|
||||
|
||||
@EventListener(condition = "#event.id.startsWith('VIP')")
|
||||
@Order(2)
|
||||
void vip(OrderPlaced event) {
|
||||
Trace.add("vip (id starts with VIP)");
|
||||
}
|
||||
|
||||
@EventListener(condition = "#event.amount >= 100 and #event.id.startsWith('VIP')")
|
||||
@Order(3)
|
||||
void both(OrderPlaced event) {
|
||||
Trace.add("both (large and VIP)");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.conditional;
|
||||
|
||||
public record OrderPlaced(String id, int amount) {
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
package com.ankurm.events.generics;
|
||||
|
||||
/** A generic event. Erasure means the JVM cannot tell Created<Order> from Created<Customer>. */
|
||||
public record Created<T>(T value) {
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.generics;
|
||||
|
||||
public record Customer(String name) {
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package com.ankurm.events.generics;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
public class GenericListeners {
|
||||
|
||||
@EventListener
|
||||
void createdOrder(Created<Order> event) {
|
||||
Trace.add("Created<Order> listener got a %s", event.value().getClass().getSimpleName());
|
||||
}
|
||||
|
||||
@EventListener
|
||||
void createdCustomer(Created<Customer> event) {
|
||||
Trace.add("Created<Customer> listener got a %s", event.value().getClass().getSimpleName());
|
||||
}
|
||||
|
||||
/** A wildcard says "any Created", which is the one form that matches a payload whose type argument is unknown. */
|
||||
@EventListener
|
||||
void anyCreated(Created<?> event) {
|
||||
Trace.add("Created<?> listener got a %s", event.value().getClass().getSimpleName());
|
||||
}
|
||||
|
||||
@EventListener
|
||||
void typedOrder(TypedCreated<Order> event) {
|
||||
Trace.add("TypedCreated<Order> listener got a %s", event.value().getClass().getSimpleName());
|
||||
}
|
||||
|
||||
@EventListener
|
||||
void typedCustomer(TypedCreated<Customer> event) {
|
||||
Trace.add("TypedCreated<Customer> listener got a %s", event.value().getClass().getSimpleName());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.events.generics;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class GenericsApp {
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.generics;
|
||||
|
||||
public record Order(String id) {
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
package com.ankurm.events.generics;
|
||||
|
||||
import org.springframework.core.ResolvableType;
|
||||
import org.springframework.core.ResolvableTypeProvider;
|
||||
|
||||
/** The same event, telling Spring what T is at run time. */
|
||||
public record TypedCreated<T>(T value) implements ResolvableTypeProvider {
|
||||
|
||||
@Override
|
||||
public ResolvableType getResolvableType() {
|
||||
return ResolvableType.forClassWithGenerics(getClass(), ResolvableType.forInstance(value));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package com.ankurm.events.ordering;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/** Sits in another bean, and the class name sorts after the other one. @Order still decides. */
|
||||
@Component
|
||||
public class EarlyListener {
|
||||
|
||||
@EventListener
|
||||
@Order(-10)
|
||||
void earliest(OrderPlaced event) {
|
||||
Trace.add("earliest (@Order(-10), a different bean)");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.ordering;
|
||||
|
||||
public record OrderPlaced(String id) {
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.events.ordering;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class OrderingApp {
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package com.ankurm.events.ordering;
|
||||
|
||||
import com.ankurm.events.support.Modes;
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
public class OrderingListeners {
|
||||
|
||||
@EventListener
|
||||
@Order(1)
|
||||
void first(OrderPlaced event) {
|
||||
Trace.add("first (@Order(1))");
|
||||
}
|
||||
|
||||
@EventListener
|
||||
@Order(2)
|
||||
void second(OrderPlaced event) {
|
||||
Trace.add("second (@Order(2))");
|
||||
if (Modes.fails("second")) {
|
||||
throw new IllegalStateException("second listener failed");
|
||||
}
|
||||
}
|
||||
|
||||
/** No @Order: sorts last. */
|
||||
@EventListener
|
||||
void unordered(OrderPlaced event) {
|
||||
Trace.add("unordered (no @Order)");
|
||||
}
|
||||
|
||||
/** A returned event is published straight away, from inside this listener's turn. */
|
||||
@EventListener
|
||||
@Order(0)
|
||||
ReceiptIssued receipt(OrderPlaced event) {
|
||||
Trace.add("receipt (@Order(0)) returns a ReceiptIssued");
|
||||
return new ReceiptIssued(event.id());
|
||||
}
|
||||
|
||||
@EventListener
|
||||
void onReceipt(ReceiptIssued event) {
|
||||
Trace.add("onReceipt received %s", event);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.ordering;
|
||||
|
||||
public record ReceiptIssued(String orderId) {
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
package com.ankurm.events.support;
|
||||
|
||||
/** Switches a scenario reads at run time so one set of beans can show both the working and the failing case. */
|
||||
public final class Modes {
|
||||
|
||||
/** Name of the listener that should throw, or {@code ""} for none. */
|
||||
public static volatile String failIn = "";
|
||||
|
||||
private Modes() {
|
||||
}
|
||||
|
||||
public static boolean fails(String listener) {
|
||||
return failIn.equals(listener);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package com.ankurm.events.support;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
|
||||
/** A tiny in-memory log the scenarios write to and the tests print, so the order of events is data and not a guess. */
|
||||
public final class Trace {
|
||||
|
||||
private static final List<String> LINES = new CopyOnWriteArrayList<>();
|
||||
|
||||
private Trace() {
|
||||
}
|
||||
|
||||
public static void add(String format, Object... args) {
|
||||
LINES.add(args.length == 0 ? format : String.format(format, args));
|
||||
}
|
||||
|
||||
/** Returns everything recorded so far and clears the log. */
|
||||
public static List<String> drain() {
|
||||
var copy = new ArrayList<>(LINES);
|
||||
LINES.clear();
|
||||
return copy;
|
||||
}
|
||||
|
||||
/** Which kind of thread is running this code, in words that stay the same from run to run. */
|
||||
public static String where() {
|
||||
Thread t = Thread.currentThread();
|
||||
if (t.isVirtual()) {
|
||||
return "a virtual thread";
|
||||
}
|
||||
return t.getName().equals("main") ? "main" : "platform thread " + t.getName();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
package com.ankurm.events.support;
|
||||
|
||||
import java.sql.SQLException;
|
||||
import javax.sql.DataSource;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* Answers "could somebody else see this row right now?" by asking on a brand-new connection. A connection the
|
||||
* pool hands out here is not the one the transaction holds, so it sees only what has been committed.
|
||||
*/
|
||||
@Component
|
||||
public class Visibility {
|
||||
|
||||
private final DataSource dataSource;
|
||||
|
||||
public Visibility(DataSource dataSource) {
|
||||
this.dataSource = dataSource;
|
||||
}
|
||||
|
||||
public boolean orderVisibleElsewhere(String id) {
|
||||
return count("orders", id) == 1;
|
||||
}
|
||||
|
||||
public int count(String table, String id) {
|
||||
try (var connection = dataSource.getConnection();
|
||||
var statement = connection.prepareStatement("select count(*) from " + table + " where id = ?")) {
|
||||
statement.setString(1, id);
|
||||
try (var rows = statement.executeQuery()) {
|
||||
rows.next();
|
||||
return rows.getInt(1);
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
package com.ankurm.events.tx;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.event.TransactionalEventListener;
|
||||
|
||||
/** Only present when a scenario asks for it, so the phase transcripts show the five phase listeners and nothing else. */
|
||||
@Component
|
||||
@ConditionalOnProperty("events.fallback")
|
||||
public class FallbackListener {
|
||||
|
||||
/** Runs after commit when there is a transaction, and immediately when there is none. */
|
||||
@TransactionalEventListener(fallbackExecution = true)
|
||||
@org.springframework.core.annotation.Order(1)
|
||||
void afterCommitOrNow(OrderPlaced event) {
|
||||
Trace.add("AFTER_COMMIT or immediately (fallbackExecution = true)");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.tx;
|
||||
|
||||
public record OrderPlaced(String id, int amount) {
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package com.ankurm.events.tx;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.jdbc.core.simple.JdbcClient;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
@Service
|
||||
public class OrderService {
|
||||
|
||||
private final JdbcClient jdbc;
|
||||
private final ApplicationEventPublisher publisher;
|
||||
|
||||
public OrderService(JdbcClient jdbc, ApplicationEventPublisher publisher) {
|
||||
this.jdbc = jdbc;
|
||||
this.publisher = publisher;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void place(String id, int amount, boolean fail) {
|
||||
jdbc.sql("insert into orders(id, amount) values (?, ?)").params(id, amount).update();
|
||||
Trace.add("OrderService.place: row inserted, publishing");
|
||||
publisher.publishEvent(new OrderPlaced(id, amount));
|
||||
if (fail) {
|
||||
throw new IllegalStateException("payment declined");
|
||||
}
|
||||
Trace.add("OrderService.place: leaving the method, commit follows");
|
||||
}
|
||||
|
||||
/** Not transactional: there is no transaction for a transactional listener to wait for. */
|
||||
public void placeWithoutTransaction(String id, int amount) {
|
||||
jdbc.sql("insert into orders(id, amount) values (?, ?)").params(id, amount).update();
|
||||
Trace.add("OrderService.placeWithoutTransaction: row inserted (auto-commit), publishing");
|
||||
publisher.publishEvent(new OrderPlaced(id, amount));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package com.ankurm.events.tx;
|
||||
|
||||
import static org.springframework.transaction.event.TransactionPhase.AFTER_COMMIT;
|
||||
import static org.springframework.transaction.event.TransactionPhase.AFTER_COMPLETION;
|
||||
import static org.springframework.transaction.event.TransactionPhase.AFTER_ROLLBACK;
|
||||
import static org.springframework.transaction.event.TransactionPhase.BEFORE_COMMIT;
|
||||
|
||||
import com.ankurm.events.support.Modes;
|
||||
import com.ankurm.events.support.Trace;
|
||||
import com.ankurm.events.support.Visibility;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import org.springframework.transaction.event.TransactionalEventListener;
|
||||
|
||||
/**
|
||||
* Each listener carries an explicit {@code @Order}: two listeners in the same phase otherwise run in whatever order
|
||||
* reflection lists the methods, and the transcripts would not be repeatable.
|
||||
*/
|
||||
@Component
|
||||
public class PhaseListeners {
|
||||
|
||||
private final Visibility visibility;
|
||||
|
||||
public PhaseListeners(Visibility visibility) {
|
||||
this.visibility = visibility;
|
||||
}
|
||||
|
||||
private void note(String name, OrderPlaced event) {
|
||||
Trace.add("%-38s tx active: %-5s row visible to other connections: %s", name,
|
||||
TransactionSynchronizationManager.isActualTransactionActive(), visibility.orderVisibleElsewhere(event.id()));
|
||||
}
|
||||
|
||||
@EventListener
|
||||
@Order(0)
|
||||
void plain(OrderPlaced event) {
|
||||
note("@EventListener", event);
|
||||
if (Modes.fails("plain")) {
|
||||
throw new IllegalStateException("plain listener failed");
|
||||
}
|
||||
}
|
||||
|
||||
@TransactionalEventListener(phase = BEFORE_COMMIT)
|
||||
@Order(1)
|
||||
void beforeCommit(OrderPlaced event) {
|
||||
note("BEFORE_COMMIT", event);
|
||||
if (Modes.fails("beforeCommit")) {
|
||||
throw new IllegalStateException("BEFORE_COMMIT listener failed");
|
||||
}
|
||||
}
|
||||
|
||||
/** The default phase. */
|
||||
@TransactionalEventListener
|
||||
@Order(2)
|
||||
void afterCommit(OrderPlaced event) {
|
||||
note("AFTER_COMMIT", event);
|
||||
if (Modes.fails("afterCommit")) {
|
||||
throw new IllegalStateException("AFTER_COMMIT listener failed");
|
||||
}
|
||||
}
|
||||
|
||||
@TransactionalEventListener(phase = AFTER_ROLLBACK)
|
||||
@Order(3)
|
||||
void afterRollback(OrderPlaced event) {
|
||||
note("AFTER_ROLLBACK", event);
|
||||
}
|
||||
|
||||
@TransactionalEventListener(phase = AFTER_COMPLETION)
|
||||
@Order(4)
|
||||
void afterCompletion(OrderPlaced event) {
|
||||
note("AFTER_COMPLETION", event);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.events.tx;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication(scanBasePackages = {"com.ankurm.events.tx", "com.ankurm.events.support"})
|
||||
public class TxApp {
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package com.ankurm.events.txwrite.newtx;
|
||||
|
||||
import com.ankurm.events.txwrite.shared.OrderPlaced;
|
||||
import org.springframework.jdbc.core.simple.JdbcClient;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.annotation.Propagation;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.transaction.event.TransactionalEventListener;
|
||||
|
||||
@Component
|
||||
public class AuditNewTx {
|
||||
|
||||
private final JdbcClient jdbc;
|
||||
|
||||
public AuditNewTx(JdbcClient jdbc) {
|
||||
this.jdbc = jdbc;
|
||||
}
|
||||
|
||||
/** The write gets a transaction of its own, which is the only kind a listener that runs after commit can have. */
|
||||
@TransactionalEventListener
|
||||
@Transactional(propagation = Propagation.REQUIRES_NEW)
|
||||
void audit(OrderPlaced event) {
|
||||
jdbc.sql("insert into audit(id, note) values (?, 'written in REQUIRES_NEW')").param(event.id()).update();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.events.txwrite.newtx;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication(scanBasePackages = {"com.ankurm.events.txwrite.shared", "com.ankurm.events.txwrite.newtx"})
|
||||
public class NewTxApp {
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package com.ankurm.events.txwrite.plain;
|
||||
|
||||
import com.ankurm.events.support.Trace;
|
||||
import com.ankurm.events.support.Visibility;
|
||||
import com.ankurm.events.txwrite.shared.OrderPlaced;
|
||||
import org.springframework.jdbc.core.simple.JdbcClient;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.event.TransactionalEventListener;
|
||||
|
||||
@Component
|
||||
public class AuditPlain {
|
||||
|
||||
private final JdbcClient jdbc;
|
||||
private final Visibility visibility;
|
||||
|
||||
public AuditPlain(JdbcClient jdbc, Visibility visibility) {
|
||||
this.jdbc = jdbc;
|
||||
this.visibility = visibility;
|
||||
}
|
||||
|
||||
/** Writes to the database after commit, with nothing else said about transactions. */
|
||||
@TransactionalEventListener
|
||||
void audit(OrderPlaced event) {
|
||||
jdbc.sql("insert into audit(id, note) values (?, 'written after commit')").param(event.id()).update();
|
||||
Trace.add("inside the listener, right after the insert: audit row visible to other connections: %s",
|
||||
visibility.count("audit", event.id()) == 1);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ankurm.events.txwrite.plain;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
/** The shared service plus this package's listener, and nothing from the sibling packages. */
|
||||
@SpringBootApplication(scanBasePackages = {"com.ankurm.events.txwrite.shared", "com.ankurm.events.txwrite.plain", "com.ankurm.events.support"})
|
||||
public class PlainApp {
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
package com.ankurm.events.txwrite.required;
|
||||
|
||||
import com.ankurm.events.txwrite.shared.OrderPlaced;
|
||||
import org.springframework.jdbc.core.simple.JdbcClient;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.transaction.event.TransactionalEventListener;
|
||||
|
||||
@Component
|
||||
public class AuditRequired {
|
||||
|
||||
private final JdbcClient jdbc;
|
||||
|
||||
public AuditRequired(JdbcClient jdbc) {
|
||||
this.jdbc = jdbc;
|
||||
}
|
||||
|
||||
/** The obvious fix for a write that goes missing: add @Transactional. */
|
||||
@TransactionalEventListener
|
||||
@Transactional
|
||||
void audit(OrderPlaced event) {
|
||||
jdbc.sql("insert into audit(id, note) values (?, 'written in REQUIRED')").param(event.id()).update();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.events.txwrite.required;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication(scanBasePackages = {"com.ankurm.events.txwrite.shared", "com.ankurm.events.txwrite.required"})
|
||||
public class RequiredApp {
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
package com.ankurm.events.txwrite.shared;
|
||||
|
||||
public record OrderPlaced(String id) {
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
package com.ankurm.events.txwrite.shared;
|
||||
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.jdbc.core.simple.JdbcClient;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
@Service
|
||||
public class OrderService {
|
||||
|
||||
private final JdbcClient jdbc;
|
||||
private final ApplicationEventPublisher publisher;
|
||||
|
||||
public OrderService(JdbcClient jdbc, ApplicationEventPublisher publisher) {
|
||||
this.jdbc = jdbc;
|
||||
this.publisher = publisher;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void place(String id) {
|
||||
jdbc.sql("insert into orders(id, amount) values (?, 1)").param(id).update();
|
||||
publisher.publishEvent(new OrderPlaced(id));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
create table if not exists orders (id varchar(40) primary key, amount int not null);
|
||||
create table if not exists audit (id varchar(40) primary key, note varchar(80) not null);
|
||||
@@ -0,0 +1,99 @@
|
||||
package com.ankurm.events;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import com.ankurm.events.asyncdemo.AsyncApp;
|
||||
import com.ankurm.events.asyncdemo.AsyncOrderService;
|
||||
import com.ankurm.events.asyncdemo.Exploding;
|
||||
import com.ankurm.events.asyncdemo.Gate;
|
||||
import com.ankurm.events.asyncdemo.OrderPlaced;
|
||||
import com.ankurm.events.support.Trace;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.springframework.boot.test.system.CapturedOutput;
|
||||
import org.springframework.boot.test.system.OutputCaptureExtension;
|
||||
|
||||
/** Post 22, section 8: listeners on other threads. */
|
||||
@ExtendWith(OutputCaptureExtension.class)
|
||||
class AsyncTests {
|
||||
|
||||
@BeforeEach
|
||||
void reset() {
|
||||
Trace.drain();
|
||||
((ch.qos.logback.classic.Logger) org.slf4j.LoggerFactory.getLogger(org.slf4j.Logger.ROOT_LOGGER_NAME))
|
||||
.setLevel(ch.qos.logback.classic.Level.INFO);
|
||||
}
|
||||
|
||||
@Test
|
||||
void anAsyncListenerRunsOnAnotherThreadAndDoesNotBlockThePublisher() throws Exception {
|
||||
try (var t = new Transcript("09-async-listener-threads.txt",
|
||||
"@Async @EventListener: the publisher moves on first; which thread the listener gets")) {
|
||||
for (boolean virtual : new boolean[] {false, true}) {
|
||||
t.section("spring.threads.virtual.enabled=" + virtual);
|
||||
var run = BootRun.run(new String[] {"spring.threads.virtual.enabled=" + virtual}, AsyncApp.class);
|
||||
var gate = run.context().getBean(Gate.class);
|
||||
gate.reset();
|
||||
Trace.add("publishing on %s", Trace.where());
|
||||
run.context().publishEvent(new OrderPlaced("A-1"));
|
||||
Trace.add("publishEvent returned, releasing the listener");
|
||||
gate.publishReturned.countDown();
|
||||
assertThat(gate.slowDone.await(5, TimeUnit.SECONDS)).isTrue();
|
||||
var lines = Trace.drain();
|
||||
lines.forEach(t::line);
|
||||
run.close();
|
||||
assertThat(lines).hasSize(3);
|
||||
assertThat(lines.get(2)).contains(virtual ? "a virtual thread" : "platform thread");
|
||||
}
|
||||
|
||||
t.section("the same listener in an application without @EnableAsync");
|
||||
var plain = BootRun.run(new String[0], com.ankurm.events.asyncplain.AsyncPlainApp.class);
|
||||
plain.context().publishEvent(new com.ankurm.events.asyncplain.OrderPlaced("A-1"));
|
||||
var plainLines = Trace.drain();
|
||||
plainLines.forEach(t::line);
|
||||
plain.close();
|
||||
assertThat(plainLines).containsExactly("@Async @EventListener: running on main");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void anAsyncListenerLeavesTheTransactionBehind() throws Exception {
|
||||
try (var t = new Transcript("10-async-and-the-transaction.txt",
|
||||
"Two async listeners for an event published inside a transaction")) {
|
||||
var run = BootRun.run(new String[] {"spring.threads.virtual.enabled=true"}, AsyncApp.class);
|
||||
var gate = run.context().getBean(Gate.class);
|
||||
gate.reset();
|
||||
run.context().getBean(AsyncOrderService.class).place("as-1");
|
||||
assertThat(gate.commitDone.await(5, TimeUnit.SECONDS)).isTrue();
|
||||
var lines = Trace.drain();
|
||||
lines.forEach(t::line);
|
||||
run.close();
|
||||
assertThat(lines).hasSize(3);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void anExceptionInAnAsyncListenerNeverReachesThePublisher(CapturedOutput output) throws Exception {
|
||||
try (var t = new Transcript("11-async-exceptions.txt",
|
||||
"An @Async listener that throws: what the publisher sees, and where the exception goes")) {
|
||||
var run = BootRun.run(new String[] {"logging.level.root=ERROR", "spring.threads.virtual.enabled=true"}, AsyncApp.class);
|
||||
String outcome = "returned normally";
|
||||
try {
|
||||
run.context().publishEvent(new Exploding("A-1"));
|
||||
} catch (RuntimeException e) {
|
||||
outcome = "threw " + e;
|
||||
}
|
||||
t.line("publishEvent %s", outcome);
|
||||
for (int i = 0; i < 100 && !output.getAll().contains("Unexpected exception occurred invoking async method"); i++) {
|
||||
Thread.sleep(50);
|
||||
}
|
||||
output.getAll().lines().filter(l -> l.contains("Unexpected exception occurred invoking async method"))
|
||||
.map(l -> l.substring(l.indexOf("Unexpected")))
|
||||
.forEach(l -> t.line("logged: %s", l));
|
||||
run.close();
|
||||
assertThat(outcome).isEqualTo("returned normally");
|
||||
assertThat(output.getAll()).contains("Unexpected exception occurred invoking async method");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,151 @@
|
||||
package com.ankurm.events;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import com.ankurm.events.basics.BasicsApp;
|
||||
import com.ankurm.events.basics.OrderService;
|
||||
import com.ankurm.events.conditional.ConditionalApp;
|
||||
import com.ankurm.events.badcondition.BadConditionApp;
|
||||
import com.ankurm.events.generics.Created;
|
||||
import com.ankurm.events.generics.Customer;
|
||||
import com.ankurm.events.generics.GenericsApp;
|
||||
import com.ankurm.events.generics.Order;
|
||||
import com.ankurm.events.generics.TypedCreated;
|
||||
import com.ankurm.events.ordering.OrderingApp;
|
||||
import com.ankurm.events.support.Modes;
|
||||
import com.ankurm.events.support.Trace;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/** Post 22, sections 1 to 4: publishing, ordering, conditions and generic events. */
|
||||
class BasicsTests {
|
||||
|
||||
@BeforeEach
|
||||
@AfterEach
|
||||
void reset() {
|
||||
Trace.drain();
|
||||
Modes.failIn = "";
|
||||
}
|
||||
|
||||
@Test
|
||||
void publishingAnEventIsAnOrdinaryBlockingMethodCall() {
|
||||
try (var t = new Transcript("01-publish-is-a-method-call.txt",
|
||||
"OrderService publishes an OrderPlaced; the listeners run before publishEvent returns, on the same thread")) {
|
||||
var run = BootRun.run(new String[0], BasicsApp.class);
|
||||
assertThat(run.started()).isTrue();
|
||||
run.context().getBean(OrderService.class).place("A-1", 42);
|
||||
var lines = Trace.drain();
|
||||
lines.forEach(t::line);
|
||||
run.close();
|
||||
assertThat(lines).hasSize(4);
|
||||
assertThat(lines.get(0)).contains("publishing on main");
|
||||
assertThat(lines.get(1)).contains("AuditListener.audit").contains("OrderPlaced[id=A-1, amount=42]").endsWith("on main");
|
||||
assertThat(lines.get(2)).contains("AuditListener.counted");
|
||||
assertThat(lines.get(3)).contains("publishEvent returned");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void orderAndChaining() {
|
||||
try (var t = new Transcript("02-order-and-chaining.txt",
|
||||
"Listener order, an event returned from a listener, and a listener that throws")) {
|
||||
var run = BootRun.run(new String[0], OrderingApp.class);
|
||||
t.section("all listeners succeed");
|
||||
run.context().publishEvent(new com.ankurm.events.ordering.OrderPlaced("A-1"));
|
||||
var ok = Trace.drain();
|
||||
ok.forEach(t::line);
|
||||
assertThat(ok).hasSize(6);
|
||||
assertThat(ok.get(0)).startsWith("earliest");
|
||||
assertThat(ok.get(1)).startsWith("receipt");
|
||||
assertThat(ok.get(2)).startsWith("onReceipt");
|
||||
assertThat(ok.get(3)).startsWith("first");
|
||||
assertThat(ok.get(4)).startsWith("second");
|
||||
assertThat(ok.get(5)).startsWith("unordered");
|
||||
|
||||
t.section("the @Order(2) listener throws");
|
||||
Modes.failIn = "second";
|
||||
Throwable thrown = null;
|
||||
try {
|
||||
run.context().publishEvent(new com.ankurm.events.ordering.OrderPlaced("A-2"));
|
||||
} catch (RuntimeException e) {
|
||||
thrown = e;
|
||||
}
|
||||
Trace.drain().forEach(t::line);
|
||||
t.line("publishEvent threw: %s: %s", thrown.getClass().getSimpleName(), thrown.getMessage());
|
||||
run.close();
|
||||
assertThat(thrown).isInstanceOf(IllegalStateException.class);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void conditions() {
|
||||
try (var t = new Transcript("03-conditional-listeners.txt",
|
||||
"@EventListener(condition = ...) evaluated against the event, and a condition that names a property that does not exist")) {
|
||||
var run = BootRun.run(new String[0], ConditionalApp.class);
|
||||
record Case(String id, int amount) {
|
||||
}
|
||||
for (var c : new Case[] {new Case("VIP-1", 150), new Case("VIP-2", 20), new Case("X-3", 500), new Case("X-4", 1)}) {
|
||||
t.section("OrderPlaced(" + c.id() + ", " + c.amount() + ")");
|
||||
run.context().publishEvent(new com.ankurm.events.conditional.OrderPlaced(c.id(), c.amount()));
|
||||
var lines = Trace.drain();
|
||||
if (lines.isEmpty()) {
|
||||
t.line("(no listener ran)");
|
||||
}
|
||||
lines.forEach(t::line);
|
||||
assertThat(lines.size()).isEqualTo(switch (c.id() + c.amount()) {
|
||||
case "VIP-1150" -> 3;
|
||||
case "VIP-220", "X-3500" -> 1;
|
||||
default -> 0;
|
||||
});
|
||||
}
|
||||
run.close();
|
||||
|
||||
t.section("a condition that says #event.amountt");
|
||||
var bad = BootRun.run(new String[0], BadConditionApp.class);
|
||||
t.line("context started: %s", bad.started());
|
||||
Throwable thrown = null;
|
||||
try {
|
||||
bad.context().publishEvent(new com.ankurm.events.badcondition.OrderPlaced("A-1", 500));
|
||||
} catch (RuntimeException e) {
|
||||
thrown = e;
|
||||
}
|
||||
t.line("publishEvent threw: %s", thrown == null ? "nothing" : BootRun.chain(thrown));
|
||||
t.line("root cause: %s", thrown == null ? "-" : BootRun.root(thrown).getMessage());
|
||||
bad.close();
|
||||
assertThat(bad.started()).isTrue();
|
||||
assertThat(thrown).isNotNull();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void genericEventsAndErasure() {
|
||||
try (var t = new Transcript("04-generic-events-and-erasure.txt",
|
||||
"Created<Order> and Created<Customer> published as plain payloads, then the same with ResolvableTypeProvider")) {
|
||||
var run = BootRun.run(new String[0], GenericsApp.class);
|
||||
var plainOrder = publish(t, run, "Created<Order>", new Created<>(new Order("o-1")));
|
||||
var plainCustomer = publish(t, run, "Created<Customer>", new Created<>(new Customer("Asha")));
|
||||
var typedOrder = publish(t, run, "TypedCreated<Order>", new TypedCreated<>(new Order("o-1")));
|
||||
var typedCustomer = publish(t, run, "TypedCreated<Customer>", new TypedCreated<>(new Customer("Asha")));
|
||||
run.close();
|
||||
assertThat(plainOrder).containsExactly("Created<?> listener got a Order");
|
||||
assertThat(plainCustomer).containsExactly("Created<?> listener got a Customer");
|
||||
assertThat(typedOrder).containsExactly("TypedCreated<Order> listener got a Order");
|
||||
assertThat(typedCustomer).containsExactly("TypedCreated<Customer> listener got a Customer");
|
||||
}
|
||||
}
|
||||
|
||||
private static java.util.List<String> publish(Transcript t, BootRun.Result run, String label, Object event) {
|
||||
t.section("publishing " + label);
|
||||
String outcome = "returned normally";
|
||||
try {
|
||||
run.context().publishEvent(event);
|
||||
} catch (RuntimeException e) {
|
||||
outcome = "threw " + BootRun.chain(e);
|
||||
}
|
||||
var lines = Trace.drain();
|
||||
lines.forEach(t::line);
|
||||
t.line("publishEvent %s", outcome);
|
||||
return lines;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package com.ankurm.events;
|
||||
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
|
||||
/** Starts a Spring Boot context without a web server and returns either the context or the failure. */
|
||||
public final class BootRun {
|
||||
|
||||
private BootRun() {
|
||||
}
|
||||
|
||||
public static Result run(String[] extraProperties, Class<?>... sources) {
|
||||
var props = new java.util.ArrayList<String>();
|
||||
props.add("spring.main.banner-mode=off");
|
||||
props.add("logging.level.root=OFF");
|
||||
props.addAll(java.util.List.of(extraProperties));
|
||||
try {
|
||||
ConfigurableApplicationContext ctx = new SpringApplicationBuilder(sources)
|
||||
.web(WebApplicationType.NONE)
|
||||
.properties(props.toArray(String[]::new))
|
||||
.run();
|
||||
return new Result(ctx, null);
|
||||
} catch (RuntimeException e) {
|
||||
return new Result(null, e);
|
||||
}
|
||||
}
|
||||
|
||||
/** Names the exception classes from the outside in: what a stack trace's "Caused by:" lines say, minus the noise. */
|
||||
public static String chain(Throwable t) {
|
||||
var names = new java.util.ArrayList<String>();
|
||||
for (Throwable c = t; c != null && !names.contains(c.getClass().getSimpleName() + c.hashCode()); c = c.getCause()) {
|
||||
names.add(c.getClass().getSimpleName());
|
||||
if (c.getCause() == c) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return String.join(" -> ", names);
|
||||
}
|
||||
|
||||
public static Throwable root(Throwable t) {
|
||||
while (t.getCause() != null && t.getCause() != t) {
|
||||
t = t.getCause();
|
||||
}
|
||||
return t;
|
||||
}
|
||||
|
||||
public record Result(ConfigurableApplicationContext context, RuntimeException failure) {
|
||||
|
||||
public boolean started() {
|
||||
return failure == null;
|
||||
}
|
||||
|
||||
public void close() {
|
||||
if (context != null) {
|
||||
context.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,158 @@
|
||||
package com.ankurm.events;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import com.ankurm.events.support.Modes;
|
||||
import com.ankurm.events.support.Trace;
|
||||
import com.ankurm.events.tx.OrderService;
|
||||
import com.ankurm.events.tx.TxApp;
|
||||
import com.ankurm.events.txwrite.newtx.NewTxApp;
|
||||
import com.ankurm.events.txwrite.plain.PlainApp;
|
||||
import com.ankurm.events.txwrite.required.RequiredApp;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.springframework.boot.test.system.CapturedOutput;
|
||||
import org.springframework.boot.test.system.OutputCaptureExtension;
|
||||
import org.springframework.jdbc.core.simple.JdbcClient;
|
||||
|
||||
/** Post 22, sections 5 to 7: what a transactional listener promises, and the ways it goes quiet. */
|
||||
@ExtendWith(OutputCaptureExtension.class)
|
||||
class TransactionTests {
|
||||
|
||||
@BeforeEach
|
||||
@AfterEach
|
||||
void reset() {
|
||||
Trace.drain();
|
||||
Modes.failIn = "";
|
||||
((ch.qos.logback.classic.Logger) org.slf4j.LoggerFactory.getLogger(org.slf4j.Logger.ROOT_LOGGER_NAME))
|
||||
.setLevel(ch.qos.logback.classic.Level.INFO);
|
||||
}
|
||||
|
||||
private static int rows(BootRun.Result run, String table, String id) {
|
||||
return run.context().getBean(JdbcClient.class).sql("select count(*) from " + table + " where id = ?")
|
||||
.param(id).query(Integer.class).single();
|
||||
}
|
||||
|
||||
@Test
|
||||
void commitAndRollbackWithEveryPhase() {
|
||||
try (var t = new Transcript("05-transaction-phases-commit-and-rollback.txt",
|
||||
"The same listeners, one publisher call that commits and one that rolls back")) {
|
||||
var run = BootRun.run(new String[0], TxApp.class);
|
||||
var service = run.context().getBean(OrderService.class);
|
||||
|
||||
t.section("place(\"ok-1\", fail = false)");
|
||||
service.place("ok-1", 10, false);
|
||||
var committed = Trace.drain();
|
||||
committed.forEach(t::line);
|
||||
|
||||
t.section("place(\"bad-1\", fail = true)");
|
||||
Throwable thrown = null;
|
||||
try {
|
||||
service.place("bad-1", 10, true);
|
||||
} catch (RuntimeException e) {
|
||||
thrown = e;
|
||||
}
|
||||
var rolledBack = Trace.drain();
|
||||
rolledBack.forEach(t::line);
|
||||
t.line("caller saw: %s: %s", thrown.getClass().getSimpleName(), thrown.getMessage());
|
||||
|
||||
t.section("rows afterwards");
|
||||
t.line("ok-1 rows : %d", rows(run, "orders", "ok-1"));
|
||||
t.line("bad-1 rows: %d", rows(run, "orders", "bad-1"));
|
||||
run.close();
|
||||
assertThat(committed).anyMatch(l -> l.startsWith("AFTER_COMMIT") && l.endsWith("true"))
|
||||
.noneMatch(l -> l.startsWith("AFTER_ROLLBACK"));
|
||||
assertThat(committed).anyMatch(l -> l.startsWith("@EventListener") && l.endsWith("false"));
|
||||
assertThat(rolledBack).anyMatch(l -> l.startsWith("AFTER_ROLLBACK"))
|
||||
.noneMatch(l -> l.startsWith("AFTER_COMMIT") || l.startsWith("BEFORE_COMMIT"));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void noTransactionMeansNoTransactionalEvent() {
|
||||
try (var t = new Transcript("06-no-transaction-drops-the-event.txt",
|
||||
"Publishing outside any transaction: which listeners run")) {
|
||||
var run = BootRun.run(new String[] {"events.fallback=true"}, TxApp.class);
|
||||
run.context().getBean(OrderService.class).placeWithoutTransaction("nt-1", 5);
|
||||
var lines = Trace.drain();
|
||||
lines.forEach(t::line);
|
||||
run.close();
|
||||
assertThat(lines).hasSize(3);
|
||||
assertThat(lines).anyMatch(l -> l.contains("fallbackExecution = true"));
|
||||
assertThat(lines).noneMatch(l -> l.contains("tx active") && !l.startsWith("@EventListener"));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void writingFromAnAfterCommitListener() {
|
||||
try (var t = new Transcript("07-after-commit-writes.txt",
|
||||
"An AFTER_COMMIT listener that inserts a row: three ways to declare it")) {
|
||||
t.section("no transaction attribute on the listener");
|
||||
var plain = BootRun.run(new String[0], PlainApp.class);
|
||||
plain.context().getBean(com.ankurm.events.txwrite.shared.OrderService.class).place("w-1");
|
||||
Trace.drain().forEach(t::line);
|
||||
t.line("after place() returned, order rows: %d", rows(plain, "orders", "w-1"));
|
||||
t.line("after place() returned, audit rows: %d", rows(plain, "audit", "w-1"));
|
||||
assertThat(rows(plain, "audit", "w-1")).isEqualTo(1);
|
||||
plain.close();
|
||||
|
||||
t.section("@Transactional(propagation = REQUIRES_NEW) on the listener");
|
||||
var newTx = BootRun.run(new String[0], NewTxApp.class);
|
||||
newTx.context().getBean(com.ankurm.events.txwrite.shared.OrderService.class).place("w-2");
|
||||
t.line("order rows: %d", rows(newTx, "orders", "w-2"));
|
||||
t.line("audit rows: %d", rows(newTx, "audit", "w-2"));
|
||||
assertThat(rows(newTx, "audit", "w-2")).isEqualTo(1);
|
||||
newTx.close();
|
||||
|
||||
t.section("@Transactional (REQUIRED, the default) on the listener");
|
||||
var required = BootRun.run(new String[0], RequiredApp.class);
|
||||
t.line("context started: %s", required.started());
|
||||
if (!required.started()) {
|
||||
t.line("failure chain: %s", BootRun.chain(required.failure()));
|
||||
t.line("root cause: %s", BootRun.root(required.failure()).getMessage());
|
||||
}
|
||||
assertThat(required.started()).isFalse();
|
||||
assertThat(BootRun.root(required.failure()).getMessage()).contains("REQUIRES_NEW or NOT_SUPPORTED");
|
||||
required.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void exceptionsInListenersAndTheTransaction(CapturedOutput output) {
|
||||
try (var t = new Transcript("08-exceptions-and-transactions.txt",
|
||||
"Which listener's exception reaches the caller, and whether the order row survives")) {
|
||||
var run = BootRun.run(new String[] {"logging.level.root=ERROR"}, TxApp.class);
|
||||
var service = run.context().getBean(OrderService.class);
|
||||
int n = 0;
|
||||
for (String failing : new String[] {"plain", "beforeCommit", "afterCommit"}) {
|
||||
String id = "ex-" + (++n);
|
||||
Modes.failIn = failing;
|
||||
Throwable thrown = null;
|
||||
try {
|
||||
service.place(id, 1, false);
|
||||
} catch (RuntimeException e) {
|
||||
thrown = e;
|
||||
}
|
||||
Trace.drain();
|
||||
t.section(failing + " listener throws");
|
||||
t.line("caller saw: %s", thrown == null ? "nothing" : thrown.getClass().getSimpleName() + ": " + thrown.getMessage());
|
||||
t.line("order row survived: %s", rows(run, "orders", id) == 1);
|
||||
assertThat(thrown != null).isEqualTo(!failing.equals("afterCommit"));
|
||||
assertThat(rows(run, "orders", id) == 1).isEqualTo(failing.equals("afterCommit"));
|
||||
if (failing.equals("afterCommit")) {
|
||||
var logged = output.getAll().lines().toList();
|
||||
for (int k = 0; k < logged.size(); k++) {
|
||||
if (logged.get(k).contains("TransactionSynchronization.afterCompletion threw exception")) {
|
||||
t.line("logged: %s", logged.get(k).substring(logged.get(k).indexOf(" : ") + 3));
|
||||
t.line("logged: %s", logged.get(k + 2));
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
run.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package com.ankurm.events;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.PrintWriter;
|
||||
import java.io.StringWriter;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
|
||||
/**
|
||||
* Writes a numbered transcript under {@code output/} and echoes it to the console.
|
||||
* Every console block quoted in the article comes out of one of these files verbatim.
|
||||
*/
|
||||
public final class Transcript implements AutoCloseable {
|
||||
|
||||
private final Path path;
|
||||
private final StringWriter buffer = new StringWriter();
|
||||
private final PrintWriter out = new PrintWriter(buffer);
|
||||
|
||||
public Transcript(String fileName, String title) {
|
||||
this.path = Path.of("output", fileName);
|
||||
out.println("# " + title);
|
||||
out.println();
|
||||
}
|
||||
|
||||
public Transcript line(String format, Object... args) {
|
||||
out.println(args.length == 0 ? format : String.format(format, args));
|
||||
return this;
|
||||
}
|
||||
|
||||
public Transcript blank() {
|
||||
out.println();
|
||||
return this;
|
||||
}
|
||||
|
||||
public Transcript section(String heading) {
|
||||
out.println();
|
||||
out.println("--- " + heading + " ---");
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
out.flush();
|
||||
// Absolute paths of whoever ran the build are environment noise, not a finding.
|
||||
String text = buffer.toString().replace(System.getProperty("user.dir"), "<core-events>");
|
||||
try {
|
||||
Files.createDirectories(path.getParent());
|
||||
Files.writeString(path, text);
|
||||
} catch (IOException e) {
|
||||
throw new IllegalStateException("could not write " + path, e);
|
||||
}
|
||||
System.out.print(text);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user