package com.battery.business.web;
|
import java.io.IOException;
|
import java.util.Map;
|
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;
|
import com.battery.business.service.ITestLiveService.LiveState;
|
/** 在 HTTP 边界托管 SSE 连接和最新状态推送。 */
|
@Component
|
public class TestLiveStream {
|
private static final Logger log = LoggerFactory.getLogger(TestLiveStream.class);
|
private final ITestLiveService live;
|
/**
|
* SSE 推送线程池显式创建并命名，便于按连接排查推送任务。
|
*
|
* <p>连接关闭后立即移除已取消的定点任务，避免取消任务在延迟队列里滞留到下一个周期。</p>
|
*/
|
private final ScheduledThreadPoolExecutor scheduler = scheduler();
|
public TestLiveStream(ITestLiveService live) { this.live = live; }
|
|
private static ScheduledThreadPoolExecutor scheduler()
|
{
|
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2,
|
BasicThreadFactory.builder().namingPattern("sse-live-pool-%d").daemon(true).build());
|
executor.setRemoveOnCancelPolicy(true);
|
return executor;
|
}
|
|
/** 建立指定测试的 SSE 连接，先推一次当前状态，随后每 3 秒推一次；连接完成、超时或出错时取消该推送任务。 */
|
public SseEmitter stream(Long testId)
|
{
|
SseEmitter emitter = createEmitter();
|
// 首帧就推不出去（例如刚被删除的测试）时不排推送任务：连接已经结束了。
|
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()
|
{
|
return new SseEmitter(0L);
|
}
|
|
private void sendAll(SseEmitter emitter, Long testId) throws IOException
|
{
|
LiveState state = live.current(testId);
|
emitter.send(SseEmitter.event().name("snapshot").data(state));
|
emitter.send(SseEmitter.event().name("device-status").data(state.sources()));
|
emitter.send(SseEmitter.event().name("reception-status").data(Map.of("testId", testId, "receptionStatus", state.receptionStatus())));
|
emitter.send(SseEmitter.event().name("test-status").data(Map.of("testId", testId, "testStatus", state.testStatus())));
|
emitter.send(SseEmitter.event().name("exception").data(state.exceptions()));
|
emitter.send(SseEmitter.event().name("heartbeat").data(Map.of("at", state.updatedAt())));
|
}
|
|
|
@PreDestroy public void close() { scheduler.shutdownNow(); }
|
}
|
