|
|
@@ -0,0 +1,139 @@
|
|
|
+package com.payment.platform.module.payment.stream;
|
|
|
+
|
|
|
+import org.junit.jupiter.api.AfterEach;
|
|
|
+import org.junit.jupiter.api.Test;
|
|
|
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
|
|
+import org.springframework.transaction.event.TransactionalEventListenerFactory;
|
|
|
+import org.springframework.transaction.support.TransactionSynchronization;
|
|
|
+import org.springframework.transaction.support.TransactionSynchronizationManager;
|
|
|
+
|
|
|
+import java.util.List;
|
|
|
+import java.util.concurrent.CopyOnWriteArrayList;
|
|
|
+
|
|
|
+import static org.junit.jupiter.api.Assertions.*;
|
|
|
+
|
|
|
+/**
|
|
|
+ * {@link OrderEventRelay} 的三条事务边界,在**真 Spring 容器 + 真事件多播器**里执行验证。
|
|
|
+ *
|
|
|
+ * <p>【为什么补这一条】2026-10-05 复核实测:本仓 25 个 {@code publisher.changed(...)} 发布点里,
|
|
|
+ * **只有 1 个**在 @Transactional 方法体内 ——
|
|
|
+ * {@code HarOrderService.syncFromUpstream}(@Transactional 在 :499,发布在 :550,是 HAR 主动
|
|
|
+ * 查上游的主路径,不是兜底)。其余 24 条(含开放 API 与收银台那两条建单腿:controller 里
|
|
|
+ * 裸 {@code orderMapper.insert} + 紧跟着发布,两处都**没有**事务)走的是
|
|
|
+ * {@code fallbackExecution} 立即投递那条路。
|
|
|
+ *
|
|
|
+ * <p>而 Task 14 Step 5 的端到端实证用的是**收银台**那条腿 ⇒ 它验的是自动提交腿的
|
|
|
+ * 「信号不早于行的可见性」,{@code AFTER_COMMIT} 的**延迟投递分支一次都没被执行过**
|
|
|
+ * (此前只有 {@link OrderEventRelayTest#theAfterCommitSemanticsArePinnedInCode} 的注解反射断言)。
|
|
|
+ * 本类把那个分支变成可执行断言。
|
|
|
+ *
|
|
|
+ * <p>【量具边界,别当它验过的】本类手工驱动 Spring 的事务同步协议
|
|
|
+ * ({@code initSynchronization + setActualTransactionActive + afterCompletion(STATUS_COMMITTED)}),
|
|
|
+ * 走的是与真事务管理器**同一套**回调(6.1.6 字节码实测:门槛是
|
|
|
+ * {@code isSynchronizationActive() && isActualTransactionActive()},提交信号走
|
|
|
+ * {@code afterCompletion(STATUS_COMMITTED)},{@code PlatformSynchronization} **不覆写**
|
|
|
+ * {@code afterCommit()})。它**不**覆盖:真 {@code DataSourceTransactionManager} 驱动回调
|
|
|
+ * 的那半边(Spring 框架自身的职责),以及真事务业务腿(HAR 主动查询回写)的端到端。
|
|
|
+ * 后者仍是缺口,已记在台账 Final 裁定里。
|
|
|
+ */
|
|
|
+class OrderEventRelayTransactionBoundaryTest {
|
|
|
+
|
|
|
+ /** 只记录不广播。与 {@link OrderEventRelayTest} 里那个同形但独立一份:两条用例互不牵动。 */
|
|
|
+ static class CapturingHub extends OrderEventHub {
|
|
|
+ final List<String> submitted = new CopyOnWriteArrayList<>();
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void submitBroadcast(String channel, Long tenantId) {
|
|
|
+ submitted.add(channel + "#" + tenantId);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private final CapturingHub hub = new CapturingHub();
|
|
|
+ private AnnotationConfigApplicationContext context;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 真容器:@TransactionalEventListener 的适配器由容器的 EventListenerMethodProcessor 生成。
|
|
|
+ *
|
|
|
+ * <p>★ 必须显式注册 {@link TransactionalEventListenerFactory}:裸 {@code AnnotationConfigApplicationContext}
|
|
|
+ * 默认只带 {@code DefaultEventListenerFactory}(真容器里另一支是 Spring Boot 的
|
|
|
+ * {@code TransactionAutoConfiguration → ProxyTransactionManagementConfiguration} 注册的)。
|
|
|
+ * 少了它,同一个注解会被装配成**普通** ApplicationListener —— 三条断言全部量到
|
|
|
+ * 「立即投递」,而这不是被测代码的行为,是量具的(2026-10-05 首跑实测:2/3 红,
|
|
|
+ * 失败信息 expected: <[]> but was: <[lsy_order#7]>)。
|
|
|
+ */
|
|
|
+ private AnnotationConfigApplicationContext startContext() {
|
|
|
+ AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
|
|
|
+ ctx.registerBean("transactionalEventListenerFactory", TransactionalEventListenerFactory.class);
|
|
|
+ ctx.registerBean("orderEventHub", OrderEventHub.class, () -> hub);
|
|
|
+ ctx.registerBean("orderEventRelay", OrderEventRelay.class, () -> new OrderEventRelay(hub));
|
|
|
+ ctx.refresh();
|
|
|
+ context = ctx;
|
|
|
+ return ctx;
|
|
|
+ }
|
|
|
+
|
|
|
+ @AfterEach
|
|
|
+ void tearDown() {
|
|
|
+ if (context != null) {
|
|
|
+ context.close(); // @PreDestroy → hub.shutdown()
|
|
|
+ }
|
|
|
+ TransactionSynchronizationManager.setActualTransactionActive(false);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void insideTransaction_deliversOnlyAfterCommit() {
|
|
|
+ AnnotationConfigApplicationContext ctx = startContext();
|
|
|
+
|
|
|
+ TransactionSynchronizationManager.initSynchronization();
|
|
|
+ TransactionSynchronizationManager.setActualTransactionActive(true);
|
|
|
+ try {
|
|
|
+ ctx.publishEvent(new OrderChangedEvent("lsy_order", 7L));
|
|
|
+
|
|
|
+ assertEquals(List.of(), hub.submitted,
|
|
|
+ "事务提交前一帧都不许投递:信号先到、库还没提交 ⇒ 页面重拉看不到新行,比不推更糟");
|
|
|
+
|
|
|
+ // 事务管理器提交后回调的是 afterCompletion(STATUS_COMMITTED),不是 afterCommit()
|
|
|
+ for (TransactionSynchronization sync : TransactionSynchronizationManager.getSynchronizations()) {
|
|
|
+ sync.afterCompletion(TransactionSynchronization.STATUS_COMMITTED);
|
|
|
+ }
|
|
|
+ } finally {
|
|
|
+ TransactionSynchronizationManager.clearSynchronization();
|
|
|
+ TransactionSynchronizationManager.setActualTransactionActive(false);
|
|
|
+ }
|
|
|
+
|
|
|
+ assertEquals(List.of("lsy_order#7"), hub.submitted,
|
|
|
+ "提交后必须投递(上面那条负断言的正对照:延迟不是丢失)");
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void outsideTransaction_deliversImmediately() {
|
|
|
+ AnnotationConfigApplicationContext ctx = startContext();
|
|
|
+ assertFalse(TransactionSynchronizationManager.isSynchronizationActive(),
|
|
|
+ "本用例的前提:当前线程没有活动事务");
|
|
|
+
|
|
|
+ ctx.publishEvent(new OrderChangedEvent("lsy_refund", 9L));
|
|
|
+
|
|
|
+ assertEquals(List.of("lsy_refund#9"), hub.submitted,
|
|
|
+ "无事务腿(25 个发布点里的其余 24 个)必须立即投递 ——"
|
|
|
+ + " fallbackExecution=false 时这些通道静默地一帧都发不出去");
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ void rollback_deliversNothing() {
|
|
|
+ AnnotationConfigApplicationContext ctx = startContext();
|
|
|
+
|
|
|
+ TransactionSynchronizationManager.initSynchronization();
|
|
|
+ TransactionSynchronizationManager.setActualTransactionActive(true);
|
|
|
+ try {
|
|
|
+ ctx.publishEvent(new OrderChangedEvent("har_order", 3L));
|
|
|
+ for (TransactionSynchronization sync : TransactionSynchronizationManager.getSynchronizations()) {
|
|
|
+ sync.afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
|
|
|
+ }
|
|
|
+ } finally {
|
|
|
+ TransactionSynchronizationManager.clearSynchronization();
|
|
|
+ TransactionSynchronizationManager.setActualTransactionActive(false);
|
|
|
+ }
|
|
|
+
|
|
|
+ assertEquals(List.of(), hub.submitted,
|
|
|
+ "回滚 = 那一行不在库里,投递信号只会让页面白拉一次");
|
|
|
+ }
|
|
|
+}
|