From 92f32acd847ea1700590bc669750f23ef499840e Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 05 九月 2026 13:39:44 +0800
Subject: [PATCH] 清理旧测试结构并贯通最终业务数据链路
---
battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java | 118 +++++++++++++++++-----------------------------------------
1 files changed, 35 insertions(+), 83 deletions(-)
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java
similarity index 63%
rename from ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java
rename to battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java
index 284bdda..ceedeb7 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java
@@ -1,34 +1,27 @@
-package com.ruoyi.system.service.impl;
+package com.battery.business.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.beans.factory.annotation.Autowired;
import org.springframework.lang.Nullable;
-import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
-import com.battery.device.simulator.BtsDevelopmentSimulator;
-import com.battery.device.simulator.BtsDevelopmentSimulator.CollectorSnapshot;
-import com.battery.device.discovery.DeviceDiscoveryCandidateStore;
-import com.battery.device.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
-import com.battery.device.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;
-import com.battery.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.ITestLiveService.TestPlanRow;
-import com.ruoyi.system.service.ITestLiveService.TestPlanSnapshot;
-import com.ruoyi.system.service.ITTestService;
-import com.ruoyi.system.mapper.TTestMapper;
+import com.battery.device.api.TestDataObservationAdapter;
+import com.battery.device.api.model.ChannelIdentity;
+import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore;
+import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
+import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;
+import com.battery.device.api.model.Measurement;
+import com.battery.business.domain.TTest;
+import com.battery.business.domain.TTestBattery;
+import com.battery.business.service.ITestLiveService;
+import com.battery.business.service.ITestLiveService.TestPlanRow;
+import com.battery.business.service.ITestLiveService.TestPlanSnapshot;
+import com.battery.business.service.ITTestService;
+import com.battery.business.mapper.TTestMapper;
/** 将设备最新快照转换为测试视角，并托管 SSE 连接生命周期。 */
@Service
@@ -36,18 +29,17 @@
{
private final ITTestService tests;
private final DeviceDiscoveryCandidateStore candidates;
- private final BtsDevelopmentSimulator simulator;
+ private final TestDataObservationAdapter adapter;
private final TTestMapper mapper;
- private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
@Autowired
public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
- @Nullable BtsDevelopmentSimulator simulator, TTestMapper mapper)
- { this.tests = tests; this.candidates = candidates; this.simulator = simulator; this.mapper = mapper; }
+ @Nullable TestDataObservationAdapter adapter, TTestMapper mapper)
+ { this.tests = tests; this.candidates = candidates; this.adapter = adapter; this.mapper = mapper; }
public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
- BtsDevelopmentSimulator simulator)
- { this(tests, candidates, simulator, null); }
+ TestDataObservationAdapter adapter)
+ { this(tests, candidates, adapter, null); }
@Override public LiveState current(Long testId)
{
@@ -83,7 +75,7 @@
private TestPlanSnapshot detectorTestPlan(TTest test)
{
- if (simulator == null) return null;
+ if (adapter == null) return null;
if (test.getBatteries() == null) return null;
for (TTestBattery battery : test.getBatteries())
{
@@ -91,9 +83,10 @@
|| battery.getBinding().getChlid() == null) continue;
try
{
- var source = "NOT_STARTED".equals(test.getReceptionStatus())
- ? simulator.btsTest(battery.getBinding().getDevid(), battery.getBinding().getChlid(), 1)
- : simulator.btsTest(battery.getBinding().getDevid(), battery.getBinding().getChlid());
+ var binding = battery.getBinding();
+ var source = adapter.testPlan(new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(),
+ binding.getDevid(), binding.getSubdevid(), binding.getChlid()));
+ if (source == null) continue;
List<TestPlanRow> rows = source.rows().stream()
.map(row -> new TestPlanRow(row.rowType(), row.stepNo(), row.cyclePosition(), row.stepName(), row.durationSeconds(),
row.voltageV(), row.executionStatus(), row.cycleCount(), row.recordCount()))
@@ -101,7 +94,7 @@
return new TestPlanSnapshot(source.currentCycleNo(), source.currentStepNo(),
source.plannedCycleCount(), rows);
}
- catch (IllegalArgumentException ignored) { /* 登记设备尚未被开发模拟器提供时保持无方案事实。 */ }
+ catch (IllegalArgumentException ignored) { /* 来源未提供方案时保持为空。 */ }
}
return null;
}
@@ -119,54 +112,12 @@
return new LiveSource(source.sourceType(), source.sourceId(), source.batteryNo(), source.deviceStatus(), Map.of());
}
- @Override 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 = 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())
+ if (Objects.equals(channel.identity().adapterCode(), binding.getAdapterCode())
+ && 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()))
@@ -208,14 +159,16 @@
private LiveSource collectorSource(TTestBattery battery)
{
- if (simulator == null) return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
+ if (adapter == null || mapper == null) return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
"OFFLINE", Map.of());
- for (CollectorSnapshot collector : simulator.snapshot().collectors()) if (collector.collectorId() == battery.getCollectorId())
+ var collector = adapter.latestCollector(mapper.selectCollectorCode(battery.getCollectorId()));
+ if (collector != null)
{
Map<String, Object> value = new LinkedHashMap<>();
value.put("source", "采集器"); value.put("sampleIntervalMs", collector.sampleIntervalMs());
// 在业务响应边界转换毫秒时间，设备时间与服务器接收时间分别保留。
- value.put("measurements", collector.latestMeasurements().stream().map(item -> {
+ value.put("measurements", collector.measurements().stream().filter(item -> battery.getCollectorChannels() != null
+ && battery.getCollectorChannels().stream().anyMatch(binding -> binding.getChannelCode().equals(item.channelCode()))).map(item -> {
Map<String, Object> measurement = new LinkedHashMap<>();
measurement.put("metricCode", item.metricCode());
measurement.put("channelCode", item.channelCode());
@@ -225,14 +178,13 @@
measurement.put("serverReceivedAt", item.serverReceivedAt().toEpochMilli());
return measurement;
}).toList());
- value.put("units", collector.latestMeasurements().stream().map(item -> item.unit()).distinct().toList());
- collector.latestMeasurements().stream().map(item -> item.serverReceivedAt()).max(Instant::compareTo)
+ value.put("units", collector.measurements().stream().map(item -> item.unit()).distinct().toList());
+ collector.measurements().stream().map(item -> item.serverReceivedAt()).max(Instant::compareTo)
.ifPresent(observedAt -> value.put("observedAt", observedAt.toEpochMilli()));
- return new LiveSource("COLLECTOR", (long) collector.collectorId(), battery.getBatteryNo(),
+ return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
collector.online() ? "ONLINE" : "OFFLINE", collector.online() ? value : Map.of());
}
return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), "OFFLINE", Map.of());
}
- @PreDestroy void shutdown() { scheduler.shutdownNow(); }
}
--
Gitblit v1.7.1