| | |
| | | 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; |
| | |
| | | /** 在 HTTP 边界托管 SSE 连接和最新状态推送。 */ |
| | | @Component |
| | | public class TestLiveStream { |
| | | private static final Logger log = LoggerFactory.getLogger(TestLiveStream.class); |
| | | private final ITestLiveService live; |
| | | /** |
| | | * SSE 推送线程池显式创建并命名,便于按连接排查推送任务。 |
| | |
| | | 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() |
| | | { |