| ruoyi-admin/src/main/java/com/battery/business/web/TestLiveStream.java | ●●●●● 补丁 | 查看 | 原始文档 | blame | 历史 | |
| ruoyi-admin/src/test/java/com/battery/business/web/TestLiveStreamTest.java | ●●●●● 补丁 | 查看 | 原始文档 | blame | 历史 |
ruoyi-admin/src/main/java/com/battery/business/web/TestLiveStream.java
@@ -4,6 +4,8 @@ import java.util.concurrent.*; import jakarta.annotation.PreDestroy; import org.apache.commons.lang3.concurrent.BasicThreadFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import com.battery.business.service.ITestLiveService; @@ -11,6 +13,7 @@ /** 在 HTTP 边界托管 SSE 连接和最新状态推送。 */ @Component public class TestLiveStream { private static final Logger log = LoggerFactory.getLogger(TestLiveStream.class); private final ITestLiveService live; /** * SSE 推送线程池显式创建并命名,便于按连接排查推送任务。 @@ -32,29 +35,42 @@ public SseEmitter stream(Long testId) { SseEmitter emitter = createEmitter(); try { sendAll(emitter, testId); } catch (IOException exception) { // 客户端刷新或离开页面会断开连接,不能再交给全局异常处理器写 SSE JSON。 emitter.complete(); return emitter; } catch (RuntimeException exception) { emitter.completeWithError(exception); return emitter; } ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> { try { sendAll(emitter, testId); } catch (IOException exception) { // 写出失败表示 SSE 客户端已断开,静默结束连接。 emitter.complete(); } catch (RuntimeException exception) { emitter.completeWithError(exception); } }, 3, 3, TimeUnit.SECONDS); // 首帧就推不出去(例如刚被删除的测试)时不排推送任务:连接已经结束了。 if (!push(emitter, testId)) return emitter; ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> push(emitter, testId), 3, 3, TimeUnit.SECONDS); emitter.onCompletion(() -> future.cancel(false)); emitter.onTimeout(() -> future.cancel(false)); emitter.onError(error -> future.cancel(false)); return emitter; } /** * 推一帧,返回连接是否还活着。 * * <p><b>这条路上任何失败都只结束连接,不能交给 MVC 的异常分发</b>:响应头已经是 * {@code text/event-stream},全局异常处理器按 JSON 写 {@code AjaxResult} 时必然失败, * 于是连锁抛出 HttpMessageNotWritableException 与安全链拒绝,每 3 秒刷一屏错。 * 而最常见的失败并不是故障:**测试被删除**(ADR-0047 允许物理删除,测试整体删除是无痕例外), * 此时订阅本来就该结束。所以这里记一条 warn(只记消息与异常类名、不带堆栈,避免反复重推刷屏) * 并关闭连接,由客户端自己重新订阅或离开页面。 */ private boolean push(SseEmitter emitter, Long testId) { try { sendAll(emitter, testId); return true; } catch (IOException disconnected) { // 写出失败表示 SSE 客户端已断开,静默结束连接。 emitter.complete(); return false; } catch (RuntimeException failure) { log.warn("实时推送结束 testId={}:{}", testId, failure.toString()); emitter.complete(); return false; } } /** 创建实时流发射器,供断连生命周期测试替换。 */ protected SseEmitter createEmitter() { ruoyi-admin/src/test/java/com/battery/business/web/TestLiveStreamTest.java
New file @@ -0,0 +1,77 @@ package com.battery.business.web; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import org.junit.jupiter.api.Test; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import com.battery.business.service.ITestLiveService; import com.ruoyi.common.exception.ServiceException; /** * SSE 推送的失败收尾:**只结束连接,不交给 MVC 的异常分发**。 * * <p>背景(线上实测):测试被删除后,订阅它的实时页还在推,`current` 抛「测试不存在」; * 原实现走 `completeWithError`,Spring 于是把异常送去全局异常处理器,而响应头已经是 * `text/event-stream`,写 `AjaxResult` JSON 必然失败——连锁抛出 HttpMessageNotWritableException * 与安全链拒绝,每 3 秒刷一屏 ERROR。 */ class TestLiveStreamTest { /** 只记录被怎么结束,不真的写出:SseEmitter 未 initialize 时 complete 是安全空操作。 */ private static final class RecordingEmitter extends SseEmitter { private boolean completed; private boolean failedWithError; private RecordingEmitter() { super(0L); } @Override public synchronized void complete() { completed = true; super.complete(); } @Override public synchronized void completeWithError(Throwable failure) { failedWithError = true; super.completeWithError(failure); } } @Test void endsTheSubscriptionInsteadOfHandingTheFailureToMvcWhenTheTestIsGone() { RecordingEmitter emitter = new RecordingEmitter(); TestLiveStream stream = stream(emitter, testId -> { throw new ServiceException("测试不存在"); }); SseEmitter returned = stream.stream(4242L); assertSame(emitter, returned); assertTrue(emitter.completed); // 关键:不走 completeWithError —— 那会把异常送进 MVC 的异常分发。 assertFalse(emitter.failedWithError); } @Test void keepsPushingWhileTheTestIsThere() { RecordingEmitter emitter = new RecordingEmitter(); TestLiveStream stream = stream(emitter, testId -> liveState(testId)); stream.stream(7L); assertFalse(emitter.completed); assertFalse(emitter.failedWithError); } private static TestLiveStream stream(SseEmitter emitter, ITestLiveService live) { return new TestLiveStream(live) { @Override protected SseEmitter createEmitter() { return emitter; } }; } private static ITestLiveService.LiveState liveState(Long testId) { return new ITestLiveService.LiveState(testId, "RUNNING", "RECEIVING", 1L, java.util.List.of(), java.util.List.of(), null); } }