完善循环拼接与首页待办真实数据接口

yeg

2026-09-09 7100c8a4f7dd3fc7d53ffead4475d7a8abc51c12

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
package com.battery.business.web;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.*;
import jakarta.annotation.PreDestroy;
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 final ITestLiveService live;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    public TestLiveStream(ITestLiveService live) { this.live = live; }
    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);
        emitter.onCompletion(() -> future.cancel(false));
        emitter.onTimeout(() -> future.cancel(false));
        emitter.onError(error -> future.cancel(false));
        return emitter;
    }
 
    /** 创建实时流发射器，供断连生命周期测试替换。 */
    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(); }
}