yeg
4 天以前 46634e427fdd43e80c017938d1aa35f25a3166ac
实时推送失败不再交给 MVC 异常分发,测试被删时只结束连接

线上现象:订阅了某个已删除测试的实时页,每 3 秒刷一屏三类 ERROR。
链路是 sendAll → live.current 抛 ServiceException("测试不存在") → emitter.completeWithError
→ Spring 把异常送去全局异常处理器 → 响应头已是 text/event-stream、写不出 AjaxResult JSON
→ HttpMessageNotWritableException,再叠一层安全链 Access Denied。

- SSE 路径上任何失败都只 complete() 并记一条 warn(不带堆栈):这条路上没有 JSON 可写,
而测试被删除(ADR-0047 允许物理删除)本来就是订阅的终点,不是服务端故障。
- 首帧就失败时不排推送任务(连接已结束)。
- 新增 TestLiveStreamTest 钉住行为:结束方式是 complete,绝不 completeWithError。

ruoyi-admin 250 项通过,后端全量 BUILD SUCCESS。
1个文件已添加
1个文件已修改
127 ■■■■■ 已修改文件
ruoyi-admin/src/main/java/com/battery/business/web/TestLiveStream.java 50 ●●●●● 补丁 | 查看 | 原始文档 | blame | 历史
ruoyi-admin/src/test/java/com/battery/business/web/TestLiveStreamTest.java 77 ●●●●● 补丁 | 查看 | 原始文档 | 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);
    }
}