From ff458109478abbc015d93d6b48a323e0d23fa94e Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期日, 30 八月 2026 01:25:56 +0800
Subject: [PATCH] 实现测试控制台双状态实时流
---
ruoyi-device/src/main/java/com/ruoyi/device/discovery/DeviceDiscoveryCandidateStore.java | 11 ++
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java | 2
ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestLiveController.java | 29 +++++++
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java | 129 ++++++++++++++++++++++++++++++++
ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestLiveControllerContractTest.java | 31 +++++++
ruoyi-system/src/main/java/com/ruoyi/system/service/ITestLiveService.java | 17 ++++
ruoyi-admin/src/main/java/com/ruoyi/config/RedisDeviceDiscoveryCandidateStore.java | 4
7 files changed, 220 insertions(+), 3 deletions(-)
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/config/RedisDeviceDiscoveryCandidateStore.java b/ruoyi-admin/src/main/java/com/ruoyi/config/RedisDeviceDiscoveryCandidateStore.java
index 2ae2407..ea09421 100644
--- a/ruoyi-admin/src/main/java/com/ruoyi/config/RedisDeviceDiscoveryCandidateStore.java
+++ b/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)
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestLiveController.java b/ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestLiveController.java
new file mode 100644
index 0000000..d163467
--- /dev/null
+++ b/ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestLiveController.java
@@ -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); }
+}
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestLiveControllerContractTest.java b/ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestLiveControllerContractTest.java
new file mode 100644
index 0000000..d2fd9e9
--- /dev/null
+++ b/ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestLiveControllerContractTest.java
@@ -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); }
+ }
+}
diff --git a/ruoyi-device/src/main/java/com/ruoyi/device/discovery/DeviceDiscoveryCandidateStore.java b/ruoyi-device/src/main/java/com/ruoyi/device/discovery/DeviceDiscoveryCandidateStore.java
index 93bbc2a..a9fe59d 100644
--- a/ruoyi-device/src/main/java/com/ruoyi/device/discovery/DeviceDiscoveryCandidateStore.java
+++ b/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,
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/ITestLiveService.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/ITestLiveService.java
new file mode 100644
index 0000000..44f88bb
--- /dev/null
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/ITestLiveService.java
@@ -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) { }
+}
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java
index 87a4392..2dfe562 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java
+++ b/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(),
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java
new file mode 100644
index 0000000..b38b100
--- /dev/null
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java
@@ -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(); }
+}
--
Gitblit v1.7.1