From acfb83068486766f63284b479d30f78e3ec17dca Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 05 九月 2026 02:24:48 +0800
Subject: [PATCH] 完成测试生命周期与接收边界
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java | 35 +++++++++++++++++++++++------------
1 files changed, 23 insertions(+), 12 deletions(-)
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java
index d1dccbb..05d1688 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java
@@ -61,9 +61,11 @@
for (Long testId : testIds)
{
TTest locked = mapper.selectTestByIdForUpdate(testId);
- if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
+ if (locked == null || !"RUNNING".equals(locked.getTestStatus())
+ || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
TTest test = tests.selectTestById(testId);
- if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue;
+ if (test == null || !"RUNNING".equals(test.getTestStatus())
+ || !"RECEIVING".equals(test.getReceptionStatus())) continue;
// 完整快照与原始事实在同一接收事务内写入；原始事实按发现频率限流，避免 100ms 开发调度重复扫描历史。
Instant lastRaw = lastRawCaptureAt.get(testId);
if (lastRaw == null || !now.toInstant().isBefore(lastRaw.plusSeconds(5)))
@@ -95,7 +97,8 @@
try { collector = simulator.collectorRealtime(battery.getCollectorId().intValue()); } catch (IllegalArgumentException ignored) { return; }
if (cabinet == null || !collector.online()) return;
if (test.getStartedAtUtc() != null && cabinet.deviceTime().isBefore(test.getStartedAtUtc().toInstant())) return;
- if (test.getUpdateTime() != null && cabinet.deviceTime().isBefore(test.getUpdateTime().toInstant())) return;
+ if (test.getReceptionBoundaryAtUtc() != null && cabinet.serverReceivedAt() != null
+ && cabinet.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())) return;
Map<String, CollectorRealtimeMeasurement> metrics = new LinkedHashMap<>();
for (CollectorRealtimeMeasurement item : collector.measurements())
if (battery.getCollectorChannels() != null && battery.getCollectorChannels().stream().anyMatch(mapping ->
@@ -109,9 +112,10 @@
if (test.getStartedAtUtc() != null && (temp.deviceTime().isBefore(test.getStartedAtUtc().toInstant())
|| pos.deviceTime().isBefore(test.getStartedAtUtc().toInstant())
|| neg.deviceTime().isBefore(test.getStartedAtUtc().toInstant()))) return;
- if (test.getUpdateTime() != null && (temp.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())
- || pos.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())
- || neg.serverReceivedAt().isBefore(test.getUpdateTime().toInstant()))) return;
+ if (test.getReceptionBoundaryAtUtc() != null
+ && (temp.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())
+ || pos.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())
+ || neg.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant()))) return;
Date slot = new Date((now.getTime() / interval) * interval);
byte[] hash = snapshotHash(cabinet, temp, pos, neg, slot.toInstant());
mapper.insertRealtimeSnapshot(test.getTestId(), binding.getBindingId(), binding.getChannelId(), battery.getCollectorId(), slot,
@@ -143,9 +147,11 @@
{
// 锁住测试行直到本轮采样写入完成，暂停/停止动作只能在本轮接收提交后生效。
TTest locked = mapper.selectTestByIdForUpdate(testId);
- if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
+ if (locked == null || !"RUNNING".equals(locked.getTestStatus())
+ || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
TTest test = tests.selectTestById(testId);
- if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue;
+ if (test == null || !"RUNNING".equals(test.getTestStatus())
+ || !"RECEIVING".equals(test.getReceptionStatus())) continue;
captureTest(test);
}
}
@@ -160,9 +166,11 @@
for (Long testId : testIds)
{
TTest locked = mapper.selectTestByIdForUpdate(testId);
- if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
+ if (locked == null || !"RUNNING".equals(locked.getTestStatus())
+ || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
TTest test = tests.selectTestById(testId);
- if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue;
+ if (test == null || !"RUNNING".equals(test.getTestStatus())
+ || !"RECEIVING".equals(test.getReceptionStatus())) continue;
for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
captureCollectorMeasurements(test, battery);
}
@@ -182,6 +190,8 @@
{
// 模拟器保留通道历史；只接收本次测试开始后产生的设备时间记录，避免顺序测试串入旧事实。
if (startedAt != null && record.deviceTime().isBefore(startedAt.toInstant())) continue;
+ if (test.getReceptionBoundaryAtUtc() != null && record.serverReceivedAt() != null
+ && record.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())) continue;
if (save(test, binding, record)) {
String source = String.join(":", java.util.Objects.toString(binding.getAdapterCode(), ""),
java.util.Objects.toString(binding.getDevtype(), ""), java.util.Objects.toString(binding.getDevid(), ""),
@@ -207,8 +217,9 @@
for (CollectorRealtimeMeasurement measurement : realtime.measurements())
{
if (startedAt != null && measurement.deviceTime().isBefore(startedAt.toInstant())) continue;
- // 继续接收只接受恢复动作之后查询到的快照；暂停期间采集器自行产生的历史不回补。
- if (test.getUpdateTime() != null && measurement.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())) continue;
+ // 继续接收只接受恢复动作之后服务器收到的采样；暂停期间采集器历史不回补。
+ if (test.getReceptionBoundaryAtUtc() != null
+ && measurement.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())) continue;
boolean mapped = battery.getCollectorChannels().stream().anyMatch(mapping ->
java.util.Objects.equals(mapping.getChannelCode(), measurement.channelCode())
&& metricMatches(mapping.getMeasurementCode(), measurement.metricCode()));
--
Gitblit v1.7.1