实现测试控制台双状态实时流

yeg

2026-08-30 ff458109478abbc015d93d6b48a323e0d23fa94e

实现测试控制台双状态实时流

新增鉴权实时状态 REST 与 SSE 事件流，保存检测柜快照测量单位并展示采集器采样时间。

 ruoyi-admin/src/main/java/com/ruoyi/config/RedisDeviceDiscoveryCandidateStore.java

@@ -13,6 +13,7 @@
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.device.model.ChannelIdentity;
import com.ruoyi.device.model.DeviceWorkStatus;
import com.ruoyi.device.model.RealtimeSnapshot;

/** 保存发现候选和实时状态；条目按 expiresAt 逻辑过期，Redis key 本身不设短 TTL。 */
@Component
@@ -138,8 +139,9 @@
        ChannelIdentity channelIdentity = new ChannelIdentity(identity.getString("adapterCode"),
                identity.getIntValue("devtype"), identity.getIntValue("devid"),
                identity.getIntValue("subdevid"), identity.getIntValue("chlid"));
        RealtimeSnapshot latestSnapshot = object.getObject("latestSnapshot", RealtimeSnapshot.class);
        return new ChannelCandidate(channelIdentity, object.getBooleanValue("online"),
                DeviceWorkStatus.valueOf(object.getString("workStatus")), requiredInstant(object, "observedAt"));
                DeviceWorkStatus.valueOf(object.getString("workStatus")), requiredInstant(object, "observedAt"), latestSnapshot);
    }

    private CollectorCandidate collectorCandidate(Object value)

 ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestLiveController.java

New file
@@ -0,0 +1,29 @@
package com.ruoyi.web.controller.system;

import org.springframework.http.MediaType;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import com.ruoyi.common.core.controller.BaseController;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.system.service.ITestLiveService;

/** 测试控制台实时 REST/SSE 入口，仅负责鉴权和协议转发。 */
@RestController
@RequestMapping("/system/test")
public class TTestLiveController extends BaseController
{
    private final ITestLiveService liveService;
    public TTestLiveController(ITestLiveService liveService) { this.liveService = liveService; }

    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping("/{testId}/live-state")
    public AjaxResult liveState(@PathVariable Long testId) { return success(liveService.current(testId)); }

    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping(value = "/{testId}/live", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter live(@PathVariable Long testId) { return liveService.stream(testId); }
}

 ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestLiveControllerContractTest.java

New file
@@ -0,0 +1,31 @@
package com.ruoyi.web.controller.system;

import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.lang.reflect.Method;
import org.junit.jupiter.api.Test;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;

/** Ticket09 实时接口契约，避免前端依赖未鉴权或 URL 携带令牌的实现。 */
class TTestLiveControllerContractTest
{
    @Test
    void liveStateAndStreamMappingsAreAuthenticatedSurface()
    {
        Method state = method("liveState", Long.class);
        assertArrayEquals(new String[] { "/{testId}/live-state" }, state.getAnnotation(GetMapping.class).value());
        Method live = method("live", Long.class);
        assertArrayEquals(new String[] { "/{testId}/live" }, live.getAnnotation(GetMapping.class).value());
        assertEquals(MediaType.TEXT_EVENT_STREAM_VALUE, live.getAnnotation(GetMapping.class).produces()[0]);
        assertTrue(live.getAnnotation(org.springframework.security.access.prepost.PreAuthorize.class).value()
                .contains("system:test:query"));
    }

    private Method method(String name, Class<?>... types)
    {
        try { return TTestLiveController.class.getDeclaredMethod(name, types); }
        catch (NoSuchMethodException exception) { throw new AssertionError(exception); }
    }
}

 ruoyi-device/src/main/java/com/ruoyi/device/discovery/DeviceDiscoveryCandidateStore.java

@@ -5,6 +5,7 @@
import java.util.Optional;
import com.ruoyi.device.model.ChannelIdentity;
import com.ruoyi.device.model.DeviceWorkStatus;
import com.ruoyi.device.model.RealtimeSnapshot;

/** 保存未登记发现候选和正式设备的实时观测快照。 */
public interface DeviceDiscoveryCandidateStore
@@ -51,7 +52,15 @@
    }

    record ChannelCandidate(ChannelIdentity identity, boolean online, DeviceWorkStatus workStatus,
            Instant observedAt) { }
            Instant observedAt, RealtimeSnapshot latestSnapshot)
    {
        /** 兼容仅有状态的旧发现候选。 */
        public ChannelCandidate(ChannelIdentity identity, boolean online, DeviceWorkStatus workStatus,
                Instant observedAt)
        {
            this(identity, online, workStatus, observedAt, null);
        }
    }

    record CollectorCandidate(String adapterCode, String deviceCode, String ip, String mac, String name,
            String manufacturer, String model, int channelCount, boolean online,

 ruoyi-system/src/main/java/com/ruoyi/system/service/ITestLiveService.java

New file
@@ -0,0 +1,17 @@
package com.ruoyi.system.service;

import java.time.Instant;
import java.util.List;
import java.util.Map;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

/** 提供测试控制台的双状态实时事实与鉴权 SSE。 */
public interface ITestLiveService
{
    LiveState current(Long testId);
    SseEmitter stream(Long testId);
    record LiveState(Long testId, String testStatus, String receptionStatus, Instant updatedAt,
            List<LiveSource> sources, List<Map<String, Object>> exceptions) { }
    record LiveSource(String sourceType, Long sourceId, String batteryNo, String deviceStatus,
            Map<String, Object> latestSnapshot) { }
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java

@@ -115,7 +115,7 @@
            RealtimeSnapshot snapshot = snapshots.get(item.identity());
            DeviceWorkStatus status = snapshot == null || snapshot.workStatus() == null
                    ? DeviceWorkStatus.UNKNOWN : snapshot.workStatus();
            return new ChannelCandidate(item.identity(), item.available(), status, observedAt);
            return new ChannelCandidate(item.identity(), item.available(), status, observedAt, snapshots.get(item.identity()));
        }).toList();
        return new CabinetCandidate(source.adapterCode(), source.adapterId(),
                source.adapterCode() + "-" + key.devtype() + "-" + key.devid(),

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java

New file
@@ -0,0 +1,129 @@
package com.ruoyi.system.service.impl;

import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import jakarta.annotation.PreDestroy;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.bts.BtsDevelopmentSimulator.CollectorSnapshot;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;
import com.ruoyi.device.model.Measurement;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
import com.ruoyi.system.service.ITestLiveService;
import com.ruoyi.system.service.ITTestService;

/** 将设备最新快照转换为测试视角，并托管 SSE 连接生命周期。 */
@Service
public class TTestLiveServiceImpl implements ITestLiveService
{
    private final ITTestService tests;
    private final DeviceDiscoveryCandidateStore candidates;
    private final BtsDevelopmentSimulator simulator;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
            BtsDevelopmentSimulator simulator)
    { this.tests = tests; this.candidates = candidates; this.simulator = simulator; }

    @Override public LiveState current(Long testId)
    {
        TTest test = tests.selectTestById(testId);
        List<CabinetCandidate> cabinets = candidates.snapshot(Instant.now()).cabinets();
        List<LiveSource> sources = new ArrayList<>();
        for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
        {
            if (battery.getBinding() != null) sources.add(cabinetSource(battery, cabinets));
            if (battery.getCollectorId() != null) sources.add(collectorSource(battery));
        }
        List<Map<String, Object>> exceptions = sources.stream().filter(source -> !"ONLINE".equals(source.deviceStatus())
                && !"WORKING".equals(source.deviceStatus()) && !"PAUSED".equals(source.deviceStatus()))
                .map(source -> Map.<String, Object>of("sourceType", source.sourceType(), "sourceId", source.sourceId(),
                        "batteryNo", source.batteryNo(), "status", source.deviceStatus(), "message", "数据来源当前不可观测"))
                .toList();
        return new LiveState(testId, test.getTestStatus(), test.getReceptionStatus(), Instant.now(), sources, exceptions);
    }

    @Override public SseEmitter stream(Long testId)
    {
        SseEmitter emitter = new SseEmitter(0L);
        try { sendAll(emitter, testId); }
        catch (RuntimeException | IOException exception) { emitter.completeWithError(exception); return emitter; }
        ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
            try { sendAll(emitter, testId); }
            catch (RuntimeException | IOException 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;
    }

    private void sendAll(SseEmitter emitter, Long testId) throws IOException
    {
        LiveState state = 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())));
    }

    private LiveSource cabinetSource(TTestBattery battery, List<CabinetCandidate> cabinets)
    {
        var binding = battery.getBinding();
        for (CabinetCandidate cabinet : cabinets) for (ChannelCandidate channel : cabinet.channels())
            if (Objects.equals(channel.identity().devtype(), binding.getDevtype())
                    && Objects.equals(channel.identity().devid(), binding.getDevid())
                    && Objects.equals(channel.identity().subdevid(), binding.getSubdevid())
                    && Objects.equals(channel.identity().chlid(), binding.getChlid()))
                return new LiveSource("CABINET_CHANNEL", binding.getChannelId(), battery.getBatteryNo(),
                        channel.online() ? channel.workStatus().name() : "OFFLINE", snapshot(channel));
        return new LiveSource("CABINET_CHANNEL", binding.getChannelId(), battery.getBatteryNo(), "UNKNOWN", Map.of());
    }

    private Map<String, Object> snapshot(ChannelCandidate channel)
    {
        Map<String, Object> value = new LinkedHashMap<>();
        if (channel.latestSnapshot() == null) return value;
        value.put("source", "检测柜");
        value.put("observedAt", channel.latestSnapshot().receivedAt());
        value.put("measurements", channel.latestSnapshot().measurements().entrySet().stream().map(entry -> {
            Measurement measurement = entry.getValue();
            return Map.of("metricCode", entry.getKey(), "value", measurement.value(), "unit", measurement.unit());
        }).toList());
        value.put("units", channel.latestSnapshot().measurements().values().stream().map(Measurement::unit).distinct().toList());
        return value;
    }

    private LiveSource collectorSource(TTestBattery battery)
    {
        for (CollectorSnapshot collector : simulator.snapshot().collectors()) if (collector.collectorId() == battery.getCollectorId())
        {
            Map<String, Object> value = new LinkedHashMap<>();
            value.put("source", "采集器"); value.put("sampleIntervalMs", collector.sampleIntervalMs());
            value.put("measurements", collector.latestMeasurements());
            value.put("units", collector.latestMeasurements().stream().map(item -> item.unit()).distinct().toList());
            collector.latestMeasurements().stream().map(item -> item.serverReceivedAt()).max(Instant::compareTo)
                    .ifPresent(observedAt -> value.put("observedAt", observedAt));
            return new LiveSource("COLLECTOR", (long) collector.collectorId(), battery.getBatteryNo(),
                    collector.online() ? "ONLINE" : "OFFLINE", value);
        }
        return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), "UNKNOWN", Map.of());
    }

    @PreDestroy void shutdown() { scheduler.shutdownNow(); }
}