From 409b5ac5a3b160eacb1fefa6a10bf97b5f6860d7 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期三, 02 九月 2026 17:41:22 +0800
Subject: [PATCH] 实现采集器实时扩展测量接收
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java | 98 +++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 98 insertions(+), 0 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 e7d8555..22f195f 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
@@ -10,6 +10,8 @@
import org.springframework.transaction.annotation.Transactional;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsHistoryRecord;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
import com.ruoyi.system.domain.TTestChannelBinding;
@@ -49,6 +51,23 @@
}
}
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void captureDevelopmentCollectorMeasurements()
+ {
+ List<Long> testIds = mapper.selectReceivingTestIds();
+ if (testIds == null) return;
+ for (Long testId : testIds)
+ {
+ TTest locked = mapper.selectTestByIdForUpdate(testId);
+ if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
+ TTest test = tests.selectTestById(testId);
+ if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue;
+ for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
+ captureCollectorMeasurements(test, battery);
+ }
+ }
+
private void captureTest(TTest test)
{
Date startedAt = test.getStartedAtUtc();
@@ -65,6 +84,62 @@
if (startedAt != null && record.deviceTime().isBefore(startedAt.toInstant())) continue;
save(test, binding, record);
}
+ }
+ }
+
+ private void captureCollectorMeasurements(TTest test, TTestBattery battery)
+ {
+ if (battery.getCollectorId() == null || battery.getCollectorChannels() == null
+ || battery.getCollectorChannels().isEmpty()) return;
+ final CollectorRealtimeSnapshot realtime;
+ try { realtime = simulator.collectorRealtime(battery.getCollectorId().intValue()); }
+ catch (IllegalArgumentException ignored) { return; }
+ if (!realtime.online()) return;
+ Date startedAt = test.getStartedAtUtc();
+ 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;
+ boolean mapped = battery.getCollectorChannels().stream().anyMatch(mapping ->
+ java.util.Objects.equals(mapping.getChannelCode(), measurement.channelCode())
+ && metricMatches(mapping.getMeasurementCode(), measurement.metricCode()));
+ if (!mapped) continue;
+ saveCollector(test, battery, measurement);
+ }
+ }
+
+ private void saveCollector(TTest test, TTestBattery battery, CollectorRealtimeMeasurement measurement)
+ {
+ byte[] hash = hash(measurement);
+ String old = mapper.selectCollectorSampleContentHash(battery.getCollectorId(), measurement.channelCode(),
+ measurement.sequenceId(), measurement.metricCode());
+ String expected = HexFormat.of().formatHex(hash);
+ if (old != null)
+ {
+ if (!old.equalsIgnoreCase(expected))
+ mapper.insertSampleConflict(test.getTestId(), battery.getBinding().getChannelId(),
+ "采集器实时接收内容冲突 collector=" + battery.getCollectorId() + ", channel="
+ + measurement.channelCode() + ", metric=" + measurement.metricCode()
+ + ", seqid=" + measurement.sequenceId());
+ return;
+ }
+ Date receivedAt = new Date();
+ int inserted = mapper.insertCollectorSample(test.getTestId(), battery.getBinding().getBindingId(), battery.getCollectorId(),
+ measurement.adapterCode(),
+ measurement.channelCode(), measurement.sequenceId(), Date.from(measurement.deviceTime()),
+ measurement.deviceTimeRaw(), receivedAt, measurement.metricCode(),
+ measurement.value(), measurement.unit(), hash);
+ // 唯一键负责并发幂等；内容冲突保留首次事实，不覆盖原始记录。
+ if (inserted == 0)
+ {
+ String actual = mapper.selectCollectorSampleContentHash(battery.getCollectorId(), measurement.channelCode(),
+ measurement.sequenceId(), measurement.metricCode());
+ if (actual != null && !actual.equalsIgnoreCase(expected))
+ mapper.insertSampleConflict(test.getTestId(), battery.getBinding().getChannelId(),
+ "采集器实时并发内容冲突 collector=" + battery.getCollectorId() + ", channel="
+ + measurement.channelCode() + ", metric=" + measurement.metricCode()
+ + ", seqid=" + measurement.sequenceId());
}
}
@@ -108,4 +183,27 @@
}
catch (Exception exception) { throw new IllegalStateException("无法计算采样内容摘要", exception); }
}
+
+ private byte[] hash(CollectorRealtimeMeasurement measurement)
+ {
+ try
+ {
+ String value = measurement.adapterCode() + "|" + measurement.collectorCode() + "|"
+ + measurement.channelCode() + "|" + measurement.sequenceId() + "|"
+ + measurement.deviceTimeRaw() + "|" + measurement.metricCode() + "|"
+ + measurement.value() + "|" + measurement.unit();
+ return MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8));
+ }
+ catch (Exception exception) { throw new IllegalStateException("无法计算采集器采样内容摘要", exception); }
+ }
+
+ private boolean metricMatches(String configured, String actual)
+ {
+ if (configured == null || actual == null) return false;
+ if (configured.equals(actual)) return true;
+ String normalized = configured.toUpperCase(java.util.Locale.ROOT);
+ return (normalized.contains("TEMPERATURE") && "TEMPERATURE".equals(actual.toUpperCase(java.util.Locale.ROOT)))
+ || (normalized.contains("POSITIVE_POTENTIAL") && "POSITIVE_POTENTIAL".equals(actual.toUpperCase(java.util.Locale.ROOT)))
+ || (normalized.contains("NEGATIVE_POTENTIAL") && "NEGATIVE_POTENTIAL".equals(actual.toUpperCase(java.util.Locale.ROOT)));
+ }
}
--
Gitblit v1.7.1