From 7ab0e426264707fb85ee38d0a9a73caf41899635 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 29 八月 2026 13:11:40 +0800
Subject: [PATCH] 实现采集器独立实时与历史采样模拟
---
ruoyi-admin/src/main/java/com/ruoyi/config/DeviceAdapterRuntimeConfig.java | 15
ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java | 73 ++++
ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorSamplingScheduler.java | 25 +
ruoyi-admin/src/main/resources/static/dev/bts-simulator.html | 118 +++++++
ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java | 87 +++++
ruoyi-admin/src/test/java/com/ruoyi/config/DeviceAdapterRuntimeConfigTest.java | 4
ruoyi-admin/src/test/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorControllerTest.java | 29 +
ruoyi-device/src/main/java/com/ruoyi/device/simulator/BtsSimulatorStateStore.java | 48 +++
ruoyi-admin/src/test/java/com/ruoyi/config/RedisStateStoreTest.java | 20 +
ruoyi-admin/src/main/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorController.java | 86 +++++
ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java | 249 ++++++++++++++++
ruoyi-device/src/test/java/com/ruoyi/device/contract/SimulatorPublicContractTest.java | 90 ++++++
12 files changed, 821 insertions(+), 23 deletions(-)
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorSamplingScheduler.java b/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorSamplingScheduler.java
new file mode 100644
index 0000000..56f3215
--- /dev/null
+++ b/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorSamplingScheduler.java
@@ -0,0 +1,25 @@
+package com.ruoyi.config;
+
+import java.time.Instant;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import com.ruoyi.device.bts.BtsDevelopmentSimulator;
+
+/** 按各采集器配置频率推进开发模拟测量，不参与正式测试数据持久化。 */
+@Component
+public class DevelopmentCollectorSamplingScheduler
+{
+ private final BtsDevelopmentSimulator simulator;
+
+ public DevelopmentCollectorSamplingScheduler(BtsDevelopmentSimulator simulator)
+ {
+ this.simulator = simulator;
+ }
+
+ /** 高频检查只在采样到期时写穿 Redis，采集器之间不要求同频。 */
+ @Scheduled(fixedDelay = 100)
+ public void sample()
+ {
+ simulator.advanceCollectors(Instant.now());
+ }
+}
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/config/DeviceAdapterRuntimeConfig.java b/ruoyi-admin/src/main/java/com/ruoyi/config/DeviceAdapterRuntimeConfig.java
index 621bda0..19af6bf 100644
--- a/ruoyi-admin/src/main/java/com/ruoyi/config/DeviceAdapterRuntimeConfig.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/config/DeviceAdapterRuntimeConfig.java
@@ -3,6 +3,7 @@
import java.io.IOException;
import java.util.List;
import org.springframework.boot.ApplicationRunner;
+import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
@@ -25,22 +26,24 @@
}
@Bean(destroyMethod = "close")
- public BtsDevelopmentSimulator btsDevelopmentSimulator(BtsSimulatorStateStore stateStore) throws IOException
+ public BtsDevelopmentSimulator btsDevelopmentSimulator(BtsSimulatorStateStore stateStore,
+ @Value("${battery.dev-simulator.port:15020}") int simulatorPort) throws IOException
{
- return new BtsDevelopmentSimulator(BtsDevelopmentSimulator.DEFAULT_PORT, stateStore);
+ return new BtsDevelopmentSimulator(simulatorPort, stateStore);
}
@Bean
- public ApplicationRunner registerDevelopmentSimulatorAdapter(DeviceAdapterRegistry registry)
+ public ApplicationRunner registerDevelopmentSimulatorAdapter(DeviceAdapterRegistry registry,
+ @Value("${battery.dev-simulator.port:15020}") int simulatorPort)
{
// 固定开发装配不进入 t_device_adapter，直接注册运行时实例。
- return arguments -> registry.synchronize(developmentSimulatorRuntimeConfig());
+ return arguments -> registry.synchronize(developmentSimulatorRuntimeConfig(simulatorPort));
}
- private AdapterInstanceConfig developmentSimulatorRuntimeConfig()
+ private AdapterInstanceConfig developmentSimulatorRuntimeConfig(int simulatorPort)
{
return new AdapterInstanceConfig(BtsDevelopmentSimulator.ADAPTER_CODE, "BTS", BtsDevelopmentSimulator.HOST,
- BtsDevelopmentSimulator.DEFAULT_PORT, BtsDevelopmentSimulator.POLL_INTERVAL_MS,
+ simulatorPort, BtsDevelopmentSimulator.POLL_INTERVAL_MS,
"Asia/Hong_Kong", true,
new AdapterCapabilities(true, true, false, false));
}
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java b/ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java
index f5fa5ae..40a5707 100644
--- a/ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java
@@ -1,5 +1,7 @@
package com.ruoyi.config;
+import java.math.BigDecimal;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -133,10 +135,67 @@
private CollectorState collectorState(Object value)
{
JSONObject object = jsonObject(value);
- return new CollectorState(requiredInteger(object, "collectorId"), object.getString("deviceCode"),
- object.getString("ip"), object.getString("mac"), object.getString("name"),
- object.getString("manufacturer"), object.getString("model"),
- requiredInteger(object, "channelCount"), object.getBooleanValue("online"));
+ int collectorId = requiredInteger(object, "collectorId");
+ String deviceCode = object.getString("deviceCode");
+ String ip = object.getString("ip");
+ String mac = object.getString("mac");
+ String name = object.getString("name");
+ String manufacturer = object.getString("manufacturer");
+ String model = object.getString("model");
+ int channelCount = requiredInteger(object, "channelCount");
+ boolean online = object.getBooleanValue("online");
+ // Ticket04 状态只包含发现配置；首次恢复时以演示默认值补齐采样状态。
+ if (!object.containsKey("sampleIntervalMs"))
+ return new CollectorState(collectorId, deviceCode, ip, mac, name, manufacturer, model,
+ channelCount, online);
+
+ List<CollectorChannelState> channels = new ArrayList<>();
+ requiredArray(object, "channels").forEach(item -> channels.add(collectorChannelState(item)));
+ List<CollectorMeasurementRecord> history = new ArrayList<>();
+ requiredArray(object, "history").forEach(item -> history.add(collectorMeasurementRecord(item)));
+ return new CollectorState(collectorId, deviceCode, ip, mac, name, manufacturer, model,
+ channelCount, online, requiredInteger(object, "sampleIntervalMs"),
+ requiredLong(object, "nextSequence"), optionalInstant(object, "lastSampledAt"),
+ channels, history, collectorHistoryFaults(object.get("historyFaults")));
+ }
+
+ private CollectorChannelState collectorChannelState(Object value)
+ {
+ JSONObject object = jsonObject(value);
+ List<CollectorMetricState> metrics = new ArrayList<>();
+ requiredArray(object, "metrics").forEach(item -> metrics.add(collectorMetricState(item)));
+ return new CollectorChannelState(object.getString("channelCode"), metrics);
+ }
+
+ private CollectorMetricState collectorMetricState(Object value)
+ {
+ JSONObject object = jsonObject(value);
+ BigDecimal measurement = object.getBigDecimal("value");
+ if (measurement == null) throw new IllegalStateException("Missing BTS simulator state field: value");
+ return new CollectorMetricState(object.getString("metricCode"), measurement, object.getString("unit"));
+ }
+
+ private CollectorMeasurementRecord collectorMeasurementRecord(Object value)
+ {
+ JSONObject object = jsonObject(value);
+ BigDecimal measurement = object.getBigDecimal("value");
+ if (measurement == null) throw new IllegalStateException("Missing BTS simulator state field: value");
+ return new CollectorMeasurementRecord(object.getString("adapterCode"), object.getString("collectorCode"),
+ object.getString("channelCode"), requiredLong(object, "sequenceId"),
+ requiredInstant(object, "deviceTime"), object.getString("deviceTimeRaw"),
+ requiredInstant(object, "serverReceivedAt"), object.getString("metricCode"),
+ measurement, object.getString("unit"));
+ }
+
+ private CollectorHistoryFaults collectorHistoryFaults(Object value)
+ {
+ if (value == null) return CollectorHistoryFaults.none();
+ JSONObject object = jsonObject(value);
+ List<Long> missing = new ArrayList<>();
+ JSONArray values = object.getJSONArray("missingSequences");
+ if (values != null) values.forEach(item -> missing.add(Long.valueOf(item.toString())));
+ return new CollectorHistoryFaults(missing, object.getLong("duplicateSequence"),
+ object.getLong("conflictingSequence"), object.getBooleanValue("outOfOrder"));
}
private ChannelState channelState(Object value)
@@ -154,4 +213,24 @@
return new ChannelTombstone(requiredInteger(object, "deviceId"), object.getString("ip"),
requiredInteger(object, "channelId"), requiredInteger(object, "remainingDiscoveries"));
}
+
+ private long requiredLong(JSONObject object, String field)
+ {
+ Long value = object.getLong(field);
+ if (value == null) throw new IllegalStateException("Missing BTS simulator state field: " + field);
+ return value;
+ }
+
+ private Instant requiredInstant(JSONObject object, String field)
+ {
+ Instant value = optionalInstant(object, field);
+ if (value == null) throw new IllegalStateException("Missing BTS simulator state field: " + field);
+ return value;
+ }
+
+ private Instant optionalInstant(JSONObject object, String field)
+ {
+ String value = object.getString(field);
+ return value == null ? null : Instant.parse(value);
+ }
}
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorController.java b/ruoyi-admin/src/main/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorController.java
index 30fc2a7..7563d63 100644
--- a/ruoyi-admin/src/main/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorController.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorController.java
@@ -1,5 +1,7 @@
package com.ruoyi.web.controller.device;
+import java.math.BigDecimal;
+import java.util.List;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.http.HttpStatus;
@@ -10,6 +12,7 @@
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.server.ResponseStatusException;
import com.ruoyi.common.annotation.Anonymous;
@@ -17,6 +20,10 @@
import com.ruoyi.device.bts.BtsDevelopmentSimulator.StateSnapshot;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.device.model.DeviceWorkStatus;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorHistoryPage;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
/** 本机临时 BTS 模拟器控制 API。 */
@Anonymous
@@ -132,6 +139,72 @@
}
}
+ /** 配置采集器采样频率和可重复的历史下载故障场景。 */
+ @PutMapping("/dev/bts-simulator/collectors/{collectorId}/sampling")
+ public StateSnapshot configureCollectorSampling(@PathVariable int collectorId,
+ @RequestBody CollectorSamplingCommand command)
+ {
+ try
+ {
+ simulator.configureCollectorSampling(collectorId, command.sampleIntervalMs(),
+ new CollectorHistoryFaults(command.missingSequences(), command.duplicateSequence(),
+ command.conflictingSequence(), command.outOfOrder()));
+ return simulator.snapshot();
+ }
+ catch (IllegalArgumentException exception)
+ {
+ throw badRequest(exception);
+ }
+ }
+
+ /** 配置稳定 S 通道的固定和动态测量项。 */
+ @PutMapping("/dev/bts-simulator/collectors/{collectorId}/channels/{channelCode}")
+ public StateSnapshot configureCollectorChannel(@PathVariable int collectorId,
+ @PathVariable String channelCode, @RequestBody CollectorChannelCommand command)
+ {
+ try
+ {
+ List<CollectorMetricState> metrics = command.metrics().stream()
+ .map(metric -> new CollectorMetricState(metric.metricCode(), metric.value(), metric.unit()))
+ .toList();
+ simulator.configureCollectorChannel(collectorId, channelCode, metrics);
+ return simulator.snapshot();
+ }
+ catch (IllegalArgumentException | NullPointerException exception)
+ {
+ throw badRequest(new IllegalArgumentException("Invalid collector channel metrics", exception));
+ }
+ }
+
+ /** 查询采集器最新实时快照；离线时保留最后观测并明确返回离线事实。 */
+ @GetMapping("/dev/bts-simulator/collectors/{collectorId}/realtime")
+ public CollectorRealtimeSnapshot collectorRealtime(@PathVariable int collectorId)
+ {
+ try
+ {
+ return simulator.collectorRealtime(collectorId);
+ }
+ catch (IllegalArgumentException exception)
+ {
+ throw badRequest(exception);
+ }
+ }
+
+ /** 按偏移量分页下载采集器模拟历史，每页最多 1000 条。 */
+ @GetMapping("/dev/bts-simulator/collectors/{collectorId}/history")
+ public CollectorHistoryPage collectorHistory(@PathVariable int collectorId,
+ @RequestParam(defaultValue = "0") long offset, @RequestParam(defaultValue = "100") int limit)
+ {
+ try
+ {
+ return simulator.collectorHistory(collectorId, offset, limit);
+ }
+ catch (IllegalArgumentException exception)
+ {
+ throw badRequest(exception);
+ }
+ }
+
/** 删除指定虚拟采集器。 */
@DeleteMapping("/dev/bts-simulator/collectors/{collectorId}")
public StateSnapshot deleteCollector(@PathVariable int collectorId)
@@ -177,4 +250,17 @@
public record CreateCollectorCommand(String ip, int channelCount) { }
public record CollectorStateCommand(String ip, int channelCount, boolean online) { }
+
+ public record CollectorSamplingCommand(int sampleIntervalMs, List<Long> missingSequences,
+ Long duplicateSequence, Long conflictingSequence, boolean outOfOrder)
+ {
+ public CollectorSamplingCommand
+ {
+ missingSequences = missingSequences == null ? List.of() : List.copyOf(missingSequences);
+ }
+ }
+
+ public record CollectorChannelCommand(List<CollectorMetricCommand> metrics) { }
+
+ public record CollectorMetricCommand(String metricCode, BigDecimal value, String unit) { }
}
diff --git a/ruoyi-admin/src/main/resources/static/dev/bts-simulator.html b/ruoyi-admin/src/main/resources/static/dev/bts-simulator.html
index 270d4c8..c2da168 100644
--- a/ruoyi-admin/src/main/resources/static/dev/bts-simulator.html
+++ b/ruoyi-admin/src/main/resources/static/dev/bts-simulator.html
@@ -25,10 +25,12 @@
.status.online::before { background: #218a5b; }
.add-form, .device-settings { display: grid; grid-template-columns: minmax(180px, 1fr) 150px auto; gap: 12px; align-items: end; padding: 16px; }
.device-settings { grid-template-columns: minmax(180px, 1fr) 130px 140px auto auto; }
- .collector-settings { display: grid; grid-template-columns: minmax(180px, 1fr) 130px 140px auto auto; gap: 12px; align-items: end; padding: 16px; }
+ .collector-settings { display: grid; grid-template-columns: minmax(160px, 1fr) 110px 130px 140px auto auto; gap: 12px; align-items: end; padding: 16px; }
+ .sampling-settings { display: grid; grid-template-columns: repeat(4, minmax(120px, 1fr)) auto; gap: 12px; align-items: end; padding: 0 16px 16px; }
label { color: #5e6a73; font-size: 12px; }
label > span { display: block; margin-bottom: 6px; }
- input[type="text"], input[type="number"], select { width: 100%; height: 34px; padding: 0 9px; color: #26323d; background: #fff; border: 1px solid #bdc8d0; border-radius: 3px; }
+ input[type="text"], input[type="number"], select, textarea { width: 100%; height: 34px; padding: 0 9px; color: #26323d; background: #fff; border: 1px solid #bdc8d0; border-radius: 3px; }
+ textarea { min-height: 64px; padding: 7px 9px; resize: vertical; font: 12px Consolas, monospace; }
input[type="checkbox"] { width: 17px; height: 17px; accent-color: #218a5b; cursor: pointer; }
.switch-label { display: inline-flex; align-items: center; gap: 8px; height: 34px; }
.switch-label > span { margin: 0; }
@@ -46,6 +48,8 @@
.request-time { width: 185px; font: 12px Consolas, monospace; color: #53616b; }
.request-command { width: 130px; font: 12px Consolas, monospace; font-weight: 600; }
.request-xml { overflow: hidden; color: #5c6871; font: 11px Consolas, monospace; text-overflow: ellipsis; white-space: nowrap; }
+ .collector-actions { display: flex; gap: 8px; padding: 12px 16px; border-top: 1px solid #e3e9ed; }
+ .collector-output { min-height: 46px; margin: 0 16px 16px; padding: 10px; overflow: auto; color: #34434d; background: #f7f9fa; border: 1px solid #d5dde3; font: 11px Consolas, monospace; white-space: pre-wrap; }
.empty { height: 96px; color: #7d8992; text-align: center; }
.error { display: none; margin-bottom: 16px; padding: 11px 14px; color: #8b2635; background: #fff0f2; border: 1px solid #e4a8b0; font-size: 13px; }
.error.visible { display: block; }
@@ -54,7 +58,7 @@
main { width: calc(100% - 20px); margin-top: 12px; }
.summary { grid-template-columns: 1fr; }
.summary > div { border-right: 0; border-bottom: 1px solid #e2e8ec; }
- .add-form, .device-settings, .collector-settings { grid-template-columns: 1fr 1fr; }
+ .add-form, .device-settings, .collector-settings, .sampling-settings { grid-template-columns: 1fr 1fr; }
.device-settings button, .collector-settings button { width: 100%; }
.device { margin: 10px; overflow-x: auto; }
.device table { min-width: 520px; }
@@ -232,6 +236,9 @@
const channelCount = field('通道数', 'number', collector.channelCount);
channelCount.input.min = '1';
channelCount.input.max = '99';
+ const sampleInterval = field('采样间隔（毫秒）', 'number', collector.sampleIntervalMs);
+ sampleInterval.input.min = '100';
+ sampleInterval.input.max = '60000';
const onlineLabel = document.createElement('label');
onlineLabel.className = 'switch-label';
const online = document.createElement('input');
@@ -245,7 +252,10 @@
save.type = 'button';
save.textContent = '保存配置';
save.addEventListener('click', async () => {
- try { render(await request(`/collectors/${collector.collectorId}`, { method: 'PUT', body: JSON.stringify({ ip: ip.input.value, channelCount: Number(channelCount.input.value), online: online.checked }) })); }
+ try {
+ await request(`/collectors/${collector.collectorId}`, { method: 'PUT', body: JSON.stringify({ ip: ip.input.value, channelCount: Number(channelCount.input.value), online: online.checked }) });
+ render(await request(`/collectors/${collector.collectorId}/sampling`, { method: 'PUT', body: JSON.stringify({ sampleIntervalMs: Number(sampleInterval.input.value), ...collector.historyFaults }) }));
+ }
catch (error) { showError(error); }
});
const remove = document.createElement('button');
@@ -256,12 +266,108 @@
try { render(await request(`/collectors/${collector.collectorId}`, { method: 'DELETE' })); }
catch (error) { showError(error); }
});
- settings.append(ip.label, channelCount.label, onlineLabel, save, remove);
- panel.append(title, settings);
+ settings.append(ip.label, channelCount.label, sampleInterval.label, onlineLabel, save, remove);
+
+ const faults = document.createElement('div');
+ faults.className = 'sampling-settings';
+ const missing = field('缺失序号（逗号分隔）', 'text', collector.historyFaults.missingSequences.join(','));
+ const duplicate = field('重复序号', 'number', collector.historyFaults.duplicateSequence || '');
+ const conflict = field('冲突序号', 'number', collector.historyFaults.conflictingSequence || '');
+ const outOfOrderLabel = document.createElement('label');
+ outOfOrderLabel.className = 'switch-label';
+ const outOfOrder = document.createElement('input');
+ outOfOrder.type = 'checkbox';
+ outOfOrder.checked = collector.historyFaults.outOfOrder;
+ const outOfOrderText = document.createElement('span');
+ outOfOrderText.textContent = '历史乱序';
+ outOfOrderLabel.append(outOfOrder, outOfOrderText);
+ const saveFaults = document.createElement('button');
+ saveFaults.type = 'button';
+ saveFaults.textContent = '保存历史场景';
+ saveFaults.addEventListener('click', async () => {
+ const missingSequences = missing.input.value.split(',').map(value => value.trim()).filter(Boolean).map(Number);
+ try { render(await request(`/collectors/${collector.collectorId}/sampling`, { method: 'PUT', body: JSON.stringify({ sampleIntervalMs: Number(sampleInterval.input.value), missingSequences, duplicateSequence: numberOrNull(duplicate.input.value), conflictingSequence: numberOrNull(conflict.input.value), outOfOrder: outOfOrder.checked }) })); }
+ catch (error) { showError(error); }
+ });
+ faults.append(missing.label, duplicate.label, conflict.label, outOfOrderLabel, saveFaults);
+
+ const table = document.createElement('table');
+ table.innerHTML = '<thead><tr><th style="width:110px">稳定通道</th><th>测量项（每行：指标编码=值 单位）</th><th style="width:130px">操作</th></tr></thead>';
+ const channelBody = document.createElement('tbody');
+ channelBody.append(...collector.channels.map(channel => collectorChannelRow(collector.collectorId, channel)));
+ table.append(channelBody);
+
+ const actions = document.createElement('div');
+ actions.className = 'collector-actions';
+ const realtimeButton = document.createElement('button');
+ realtimeButton.type = 'button';
+ realtimeButton.textContent = '查看实时';
+ const historyButton = document.createElement('button');
+ historyButton.type = 'button';
+ historyButton.textContent = '下载历史';
+ const historyCount = document.createElement('span');
+ historyCount.className = 'label';
+ historyCount.textContent = `原始历史 ${collector.historyRecordCount} 条`;
+ const output = document.createElement('pre');
+ output.className = 'collector-output';
+ output.textContent = collector.latestMeasurements.length
+ ? measurementLines(collector.latestMeasurements)
+ : '等待第一条采样';
+ realtimeButton.addEventListener('click', async () => {
+ try { output.textContent = measurementLines((await request(`/collectors/${collector.collectorId}/realtime`)).measurements); }
+ catch (error) { showError(error); }
+ });
+ historyButton.addEventListener('click', async () => {
+ try {
+ const page = await request(`/collectors/${collector.collectorId}/history?offset=0&limit=20`);
+ output.textContent = `历史 ${page.total} 条 / 本页 ${page.records.length} 条\n${measurementLines(page.records)}`;
+ } catch (error) { showError(error); }
+ });
+ actions.append(realtimeButton, historyButton, historyCount);
+ panel.append(title, settings, faults, table, actions, output);
return panel;
}));
}
+ function collectorChannelRow(collectorId, channel) {
+ const row = document.createElement('tr');
+ const identity = document.createElement('td');
+ identity.textContent = channel.channelCode;
+ const metricsCell = document.createElement('td');
+ const metrics = document.createElement('textarea');
+ metrics.setAttribute('aria-label', `${channel.channelCode} 测量项`);
+ metrics.value = channel.metrics.map(item => `${item.metricCode}=${item.value} ${item.unit}`).join('\n');
+ metricsCell.append(metrics);
+ const action = document.createElement('td');
+ const save = document.createElement('button');
+ save.type = 'button';
+ save.textContent = '保存测量值';
+ save.addEventListener('click', async () => {
+ try { render(await request(`/collectors/${collectorId}/channels/${channel.channelCode}`, { method: 'PUT', body: JSON.stringify({ metrics: parseMetrics(metrics.value) }) })); }
+ catch (error) { showError(error); }
+ });
+ action.append(save);
+ row.append(identity, metricsCell, action);
+ return row;
+ }
+
+ function parseMetrics(text) {
+ return text.split('\n').map(line => line.trim()).filter(Boolean).map(line => {
+ const match = line.match(/^([^=\s]+)=(-?\d+(?:\.\d+)?)\s+(.+)$/);
+ if (!match) throw new Error(`测量项格式错误：${line}`);
+ return { metricCode: match[1], value: Number(match[2]), unit: match[3] };
+ });
+ }
+
+ function measurementLines(items) {
+ if (!items.length) return '暂无测量';
+ return items.map(item => `${item.collectorCode} #${item.sequenceId} ${item.channelCode} ${item.metricCode}=${item.value} ${item.unit} 设备 ${item.deviceTimeRaw} / 接收 ${item.serverReceivedAt}`).join('\n');
+ }
+
+ function numberOrNull(value) {
+ return value.trim() ? Number(value) : null;
+ }
+
function field(labelText, type, value) {
const label = document.createElement('label');
const text = document.createElement('span');
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/config/DeviceAdapterRuntimeConfigTest.java b/ruoyi-admin/src/test/java/com/ruoyi/config/DeviceAdapterRuntimeConfigTest.java
index cbceb70..f16a9b2 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/config/DeviceAdapterRuntimeConfigTest.java
+++ b/ruoyi-admin/src/test/java/com/ruoyi/config/DeviceAdapterRuntimeConfigTest.java
@@ -15,7 +15,7 @@
void registersDevelopmentSimulatorBeforeApplicationReady() throws Exception
{
DeviceAdapterRegistry registry = mock(DeviceAdapterRegistry.class);
- var runner = new DeviceAdapterRuntimeConfig().registerDevelopmentSimulatorAdapter(registry);
+ var runner = new DeviceAdapterRuntimeConfig().registerDevelopmentSimulatorAdapter(registry, 15021);
runner.run(new DefaultApplicationArguments(new String[0]));
@@ -23,7 +23,7 @@
verify(registry).synchronize(adapter.capture());
assertEquals("dev-sim", adapter.getValue().adapterCode());
assertEquals("127.0.0.1", adapter.getValue().host());
- assertEquals(15020, adapter.getValue().port());
+ assertEquals(15021, adapter.getValue().port());
assertEquals(5000, adapter.getValue().pollIntervalMs());
}
}
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/config/RedisStateStoreTest.java b/ruoyi-admin/src/test/java/com/ruoyi/config/RedisStateStoreTest.java
index 94f9d19..6ccc48d 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/config/RedisStateStoreTest.java
+++ b/ruoyi-admin/src/test/java/com/ruoyi/config/RedisStateStoreTest.java
@@ -62,6 +62,26 @@
}
@Test
+ void upgradesLegacyCollectorDiscoveryStateWithSamplingDefaults()
+ {
+ AtomicReference<Object> stored = new AtomicReference<>("{\"version\":1,\"devices\":[],"
+ + "\"collectors\":[{\"collectorId\":1,\"deviceCode\":\"dev-sim-collector-1\","
+ + "\"ip\":\"127.0.1.2\",\"mac\":\"02:43:00:00:00:01\","
+ + "\"name\":\"模拟采集器\",\"manufacturer\":\"开发模拟器\","
+ + "\"model\":\"Virtual Collector\",\"channelCount\":2,\"online\":true}],"
+ + "\"nextDeviceId\":1,\"nextCollectorId\":2,\"tombstones\":[]}");
+
+ SimulatorState restored = store(stored).loadOrInitialize(state(1, "127.0.0.2"));
+
+ CollectorState collector = restored.collectors().get(0);
+ assertEquals(1000, collector.sampleIntervalMs());
+ assertEquals(1L, collector.nextSequence());
+ assertEquals(List.of("S-01", "S-02"),
+ collector.channels().stream().map(channel -> channel.channelCode()).toList());
+ assertEquals(List.of(), collector.history());
+ }
+
+ @Test
void rejectsIncompatibleVersionWithoutOverwritingIt()
{
AtomicReference<Object> stored = new AtomicReference<>("{\"version\":2,\"devices\":[],"
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorControllerTest.java b/ruoyi-admin/src/test/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorControllerTest.java
index 518fba3..720b47c 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorControllerTest.java
+++ b/ruoyi-admin/src/test/java/com/ruoyi/web/controller/device/BtsDevelopmentSimulatorControllerTest.java
@@ -6,6 +6,7 @@
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import java.lang.reflect.Method;
+import java.math.BigDecimal;
import java.net.InetAddress;
import java.time.Instant;
import java.util.List;
@@ -95,6 +96,34 @@
}
@Test
+ void controllerExposesCollectorRealtimeAndPagedHistory() throws Exception
+ {
+ try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0))
+ {
+ BtsDevelopmentSimulatorController controller = controller(simulator);
+ controller.configureCollectorSampling(1,
+ new BtsDevelopmentSimulatorController.CollectorSamplingCommand(
+ 100, List.of(), null, null, false));
+ controller.configureCollectorChannel(1, "S-01",
+ new BtsDevelopmentSimulatorController.CollectorChannelCommand(List.of(
+ new BtsDevelopmentSimulatorController.CollectorMetricCommand(
+ "temperature", new BigDecimal("27.2"), "℃"),
+ new BtsDevelopmentSimulatorController.CollectorMetricCommand(
+ "custom_voltage", new BigDecimal("1.234"), "V"))));
+ simulator.advanceCollectors(Instant.parse("2026-08-29T01:00:00Z"));
+
+ var realtime = controller.collectorRealtime(1);
+ var firstPage = controller.collectorHistory(1, 0, 2);
+
+ assertTrue(realtime.online());
+ assertTrue(realtime.measurements().stream().anyMatch(item -> item.metricCode().equals("custom_voltage")));
+ assertEquals(2, firstPage.records().size());
+ assertTrue(firstPage.hasMore());
+ assertEquals(2, firstPage.nextOffset());
+ }
+ }
+
+ @Test
void virtualIpMustBeUniqueAcrossCabinetsAndCollectors() throws Exception
{
try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0))
diff --git a/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java b/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java
index c299502..ddac794 100644
--- a/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java
+++ b/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java
@@ -12,11 +12,18 @@
import com.ruoyi.device.simulator.BtsSimulatorStateStore;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelTombstone;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorChannelState;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.DeviceState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.RecentRequest;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;
import com.ruoyi.device.simulator.InMemoryBtsSimulatorStateStore;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorHistoryPage;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
/** 提供由开发后端托管的临时只读 BTS 模拟端点。 */
public final class BtsDevelopmentSimulator implements AutoCloseable
@@ -138,8 +145,7 @@
CollectorState current = requireCollector(collectorId);
List<CollectorState> collectors = new ArrayList<>(state.collectors());
collectors.replaceAll(item -> item.collectorId() == collectorId
- ? new CollectorState(collectorId, current.deviceCode(), ip.trim(), current.mac(), current.name(),
- current.manufacturer(), current.model(), channelCount, online)
+ ? DevelopmentCollectorSimulation.resize(current, ip, channelCount, online, Instant.now())
: item);
persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
state.nextCollectorId(), state.tombstones()));
@@ -155,6 +161,45 @@
state.nextCollectorId(), state.tombstones()));
}
+ /** 配置采集器独立采样频率和历史下载故障场景。 */
+ public synchronized void configureCollectorSampling(int collectorId, int sampleIntervalMs,
+ CollectorHistoryFaults historyFaults)
+ {
+ CollectorState current = requireCollector(collectorId);
+ replaceCollector(DevelopmentCollectorSimulation.configureSampling(current, sampleIntervalMs, historyFaults));
+ }
+
+ /** 配置一个稳定 S 通道的固定或动态演示指标。 */
+ public synchronized void configureCollectorChannel(int collectorId, String channelCode,
+ List<CollectorMetricState> metrics)
+ {
+ CollectorState current = requireCollector(collectorId);
+ replaceCollector(DevelopmentCollectorSimulation.configureChannel(current, channelCode, metrics));
+ }
+
+ /** 按每台采集器自己的频率推进实时和历史采样。 */
+ public synchronized void advanceCollectors(Instant now)
+ {
+ List<CollectorState> collectors = state.collectors().stream()
+ .map(collector -> DevelopmentCollectorSimulation.advance(collector, now))
+ .toList();
+ if (!collectors.equals(state.collectors()))
+ persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
+ state.nextCollectorId(), state.tombstones()));
+ }
+
+ /** 返回采集器最新实时快照；快照与分页历史保持不同读取路径。 */
+ public synchronized CollectorRealtimeSnapshot collectorRealtime(int collectorId)
+ {
+ return DevelopmentCollectorSimulation.realtime(requireCollector(collectorId));
+ }
+
+ /** 分页下载采集器模拟历史，并应用已配置的重复、冲突、乱序或缺失场景。 */
+ public synchronized CollectorHistoryPage collectorHistory(int collectorId, long offset, int limit)
+ {
+ return DevelopmentCollectorSimulation.history(requireCollector(collectorId), offset, limit);
+ }
+
/** 返回控制页需要的运行状态快照。 */
public synchronized StateSnapshot snapshot()
{
@@ -168,7 +213,9 @@
List<CollectorSnapshot> collectorSnapshots = state.collectors().stream()
.map(collector -> new CollectorSnapshot(collector.collectorId(), collector.deviceCode(),
collector.ip(), collector.mac(), collector.name(), collector.manufacturer(),
- collector.model(), collector.channelCount(), collector.online()))
+ collector.model(), collector.channelCount(), collector.online(), collector.sampleIntervalMs(),
+ collector.channels(), collector.historyFaults(), collector.history().size(),
+ DevelopmentCollectorSimulation.realtime(collector).measurements()))
.toList();
return new StateSnapshot(running, HOST + ":" + port(), deviceSnapshots, collectorSnapshots,
stateStore.loadRecentRequests());
@@ -202,6 +249,14 @@
return new CollectorState(collectorId, "dev-sim-collector-" + collectorId,
ip.trim(), simulatorCollectorMac(collectorId), "dev-sim 采集器 " + collectorId,
"开发模拟器", "Virtual Collector", channelCount, true);
+ }
+
+ private void replaceCollector(CollectorState replacement)
+ {
+ List<CollectorState> collectors = new ArrayList<>(state.collectors());
+ collectors.replaceAll(item -> item.collectorId() == replacement.collectorId() ? replacement : item);
+ persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
+ state.nextCollectorId(), state.tombstones()));
}
private String respond(String request)
@@ -442,6 +497,7 @@
|| !isIpv4(collector.ip()) || !ips.add(collector.ip())
|| collector.channelCount() < 1 || collector.channelCount() > MAX_CHANNEL_COUNT)
throw new IllegalStateException("Invalid BTS simulator collector state in Redis");
+ DevelopmentCollectorSimulation.validate(collector);
maximumCollectorId = Math.max(maximumCollectorId, collector.collectorId());
}
java.util.Set<String> tombstoneIds = new java.util.HashSet<>();
@@ -471,6 +527,15 @@
public record ChannelSnapshot(int channelId, boolean online, DeviceWorkStatus workStatus) { }
public record CollectorSnapshot(int collectorId, String deviceCode, String ip, String mac, String name,
- String manufacturer, String model, int channelCount, boolean online) { }
+ String manufacturer, String model, int channelCount, boolean online, int sampleIntervalMs,
+ List<CollectorChannelState> channels, CollectorHistoryFaults historyFaults, int historyRecordCount,
+ List<CollectorMeasurementRecord> latestMeasurements)
+ {
+ public CollectorSnapshot
+ {
+ channels = List.copyOf(channels);
+ latestMeasurements = List.copyOf(latestMeasurements);
+ }
+ }
}
diff --git a/ruoyi-device/src/main/java/com/ruoyi/device/simulator/BtsSimulatorStateStore.java b/ruoyi-device/src/main/java/com/ruoyi/device/simulator/BtsSimulatorStateStore.java
index c1525db..1902705 100644
--- a/ruoyi-device/src/main/java/com/ruoyi/device/simulator/BtsSimulatorStateStore.java
+++ b/ruoyi-device/src/main/java/com/ruoyi/device/simulator/BtsSimulatorStateStore.java
@@ -1,5 +1,6 @@
package com.ruoyi.device.simulator;
+import java.math.BigDecimal;
import java.time.Instant;
import java.util.List;
import com.ruoyi.device.model.DeviceWorkStatus;
@@ -43,7 +44,52 @@
}
record CollectorState(int collectorId, String deviceCode, String ip, String mac, String name,
- String manufacturer, String model, int channelCount, boolean online) { }
+ String manufacturer, String model, int channelCount, boolean online, int sampleIntervalMs,
+ long nextSequence, Instant lastSampledAt, List<CollectorChannelState> channels,
+ List<CollectorMeasurementRecord> history, CollectorHistoryFaults historyFaults)
+ {
+ public CollectorState
+ {
+ channels = List.copyOf(channels);
+ history = List.copyOf(history);
+ historyFaults = historyFaults == null ? CollectorHistoryFaults.none() : historyFaults;
+ }
+
+ /** 兼容只包含发现配置的旧模拟状态；Redis 读取边界会补齐采样合同。 */
+ public CollectorState(int collectorId, String deviceCode, String ip, String mac, String name,
+ String manufacturer, String model, int channelCount, boolean online)
+ {
+ this(collectorId, deviceCode, ip, mac, name, manufacturer, model, channelCount, online,
+ DevelopmentCollectorSimulation.DEFAULT_SAMPLE_INTERVAL_MS, 1L, null,
+ DevelopmentCollectorSimulation.defaultChannels(channelCount), List.of(),
+ CollectorHistoryFaults.none());
+ }
+ }
+
+ record CollectorChannelState(String channelCode, List<CollectorMetricState> metrics)
+ {
+ public CollectorChannelState { metrics = List.copyOf(metrics); }
+ }
+
+ record CollectorMetricState(String metricCode, BigDecimal value, String unit) { }
+
+ record CollectorMeasurementRecord(String adapterCode, String collectorCode, String channelCode,
+ long sequenceId, Instant deviceTime, String deviceTimeRaw, Instant serverReceivedAt,
+ String metricCode, BigDecimal value, String unit) { }
+
+ record CollectorHistoryFaults(List<Long> missingSequences, Long duplicateSequence,
+ Long conflictingSequence, boolean outOfOrder)
+ {
+ public CollectorHistoryFaults
+ {
+ missingSequences = missingSequences == null ? List.of() : List.copyOf(missingSequences);
+ }
+
+ public static CollectorHistoryFaults none()
+ {
+ return new CollectorHistoryFaults(List.of(), null, null, false);
+ }
+ }
record ChannelState(int channelId, boolean online, DeviceWorkStatus workStatus) { }
diff --git a/ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java b/ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java
new file mode 100644
index 0000000..c1092c5
--- /dev/null
+++ b/ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java
@@ -0,0 +1,249 @@
+package com.ruoyi.device.simulator;
+
+import java.math.BigDecimal;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorChannelState;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState;
+
+/** 实现第一版开发采集器的独立采样与故障注入合同。 */
+public final class DevelopmentCollectorSimulation
+{
+ public static final int DEFAULT_SAMPLE_INTERVAL_MS = 1000;
+ public static final int MIN_SAMPLE_INTERVAL_MS = 100;
+ public static final int MAX_SAMPLE_INTERVAL_MS = 60000;
+ public static final int MAX_HISTORY_PAGE_SIZE = 1000;
+ private static final int MAX_HISTORY_RECORDS = 50000;
+ private static final int MAX_SAMPLES_PER_ADVANCE = 1000;
+ private static final ZoneId DEVICE_ZONE = ZoneId.of("Asia/Hong_Kong");
+ private static final DateTimeFormatter DEVICE_TIME_FORMATTER =
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
+
+ private DevelopmentCollectorSimulation() { }
+
+ /** 创建带稳定 S 通道和三项固定演示指标的采集器状态。 */
+ public static List<CollectorChannelState> defaultChannels(int channelCount)
+ {
+ List<CollectorChannelState> channels = new ArrayList<>(channelCount);
+ for (int index = 1; index <= channelCount; index++)
+ {
+ channels.add(new CollectorChannelState(channelCode(index), List.of(
+ new CollectorMetricState("temperature", new BigDecimal("25.0")
+ .add(BigDecimal.valueOf(index, 1)), "℃"),
+ new CollectorMetricState("positive_potential", new BigDecimal("3.700"), "V"),
+ new CollectorMetricState("negative_potential", new BigDecimal("-0.100"), "V"))));
+ }
+ return List.copyOf(channels);
+ }
+
+ /** 调整通道数时保留已有通道配置，并为新增通道生成稳定默认配置。 */
+ public static CollectorState resize(CollectorState current, String ip, int channelCount, boolean online,
+ Instant changedAt)
+ {
+ List<CollectorChannelState> defaults = defaultChannels(channelCount);
+ List<CollectorChannelState> channels = new ArrayList<>(channelCount);
+ for (int index = 0; index < channelCount; index++)
+ channels.add(index < current.channels().size() ? current.channels().get(index) : defaults.get(index));
+ Instant lastSampledAt = online == current.online() ? current.lastSampledAt() : changedAt;
+ return copy(current, ip, channelCount, online, current.sampleIntervalMs(), current.nextSequence(),
+ lastSampledAt, channels, current.history(), current.historyFaults());
+ }
+
+ /** 更新采样频率和历史下载故障；测量原始历史保持只追加。 */
+ public static CollectorState configureSampling(CollectorState current, int sampleIntervalMs,
+ CollectorHistoryFaults historyFaults)
+ {
+ validateSampleInterval(sampleIntervalMs);
+ validateFaults(historyFaults);
+ return copy(current, current.ip(), current.channelCount(), current.online(), sampleIntervalMs,
+ current.nextSequence(), current.lastSampledAt(), current.channels(), current.history(), historyFaults);
+ }
+
+ /** 替换一个演示通道的指标配置；动态添加项与固定指标共用指标编码合同。 */
+ public static CollectorState configureChannel(CollectorState current, String channelCode,
+ List<CollectorMetricState> metrics)
+ {
+ validateMetrics(metrics);
+ int channelIndex = parseChannelIndex(channelCode, current.channelCount());
+ List<CollectorChannelState> channels = new ArrayList<>(current.channels());
+ channels.set(channelIndex, new CollectorChannelState(channelCode(channelIndex + 1), metrics));
+ return copy(current, current.ip(), current.channelCount(), current.online(), current.sampleIntervalMs(),
+ current.nextSequence(), current.lastSampledAt(), channels, current.history(), current.historyFaults());
+ }
+
+ /** 按采集器自身频率推进采样；离线期间不生成记录，恢复后从下一序号继续。 */
+ public static CollectorState advance(CollectorState current, Instant now)
+ {
+ if (!current.online()) return current;
+ Instant sampleAt = current.lastSampledAt() == null
+ ? now
+ : current.lastSampledAt().plusMillis(current.sampleIntervalMs());
+ if (sampleAt.isAfter(now)) return current;
+
+ List<CollectorMeasurementRecord> history = new ArrayList<>(current.history());
+ long sequence = current.nextSequence();
+ int generated = 0;
+ Instant lastSampledAt = current.lastSampledAt();
+ while (!sampleAt.isAfter(now) && generated < MAX_SAMPLES_PER_ADVANCE)
+ {
+ for (CollectorChannelState channel : current.channels())
+ {
+ for (CollectorMetricState metric : channel.metrics())
+ {
+ history.add(new CollectorMeasurementRecord("dev-sim", current.deviceCode(),
+ channel.channelCode(), sequence, sampleAt,
+ DEVICE_TIME_FORMATTER.format(sampleAt.atZone(DEVICE_ZONE)), now,
+ metric.metricCode(), metric.value(), metric.unit()));
+ }
+ }
+ lastSampledAt = sampleAt;
+ sequence++;
+ generated++;
+ sampleAt = sampleAt.plusMillis(current.sampleIntervalMs());
+ }
+ if (history.size() > MAX_HISTORY_RECORDS)
+ history = new ArrayList<>(history.subList(history.size() - MAX_HISTORY_RECORDS, history.size()));
+ return copy(current, current.ip(), current.channelCount(), current.online(), current.sampleIntervalMs(),
+ sequence, lastSampledAt, current.channels(), history, current.historyFaults());
+ }
+
+ /** 返回每个通道和指标的最新实时快照，不将快照写入另一条历史路径。 */
+ public static CollectorRealtimeSnapshot realtime(CollectorState current)
+ {
+ Map<String, CollectorMeasurementRecord> latest = new LinkedHashMap<>();
+ for (CollectorMeasurementRecord record : current.history())
+ latest.put(record.channelCode() + "\u0000" + record.metricCode(), record);
+ return new CollectorRealtimeSnapshot(current.deviceCode(), current.online(), current.sampleIntervalMs(),
+ List.copyOf(latest.values()));
+ }
+
+ /** 对只追加原始模拟历史应用可重复的下载故障视图，再按偏移量分页。 */
+ public static CollectorHistoryPage history(CollectorState current, long offset, int limit)
+ {
+ if (offset < 0) throw new IllegalArgumentException("offset must not be negative");
+ if (limit < 1 || limit > MAX_HISTORY_PAGE_SIZE)
+ throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_PAGE_SIZE);
+
+ List<CollectorMeasurementRecord> download = faultedHistory(current);
+ int from = (int) Math.min(offset, download.size());
+ int to = Math.min(from + limit, download.size());
+ return new CollectorHistoryPage(current.deviceCode(), offset, to, to < download.size(),
+ download.size(), download.subList(from, to));
+ }
+
+ /** 校验 Redis 恢复的采集器采样状态，拒绝损坏状态被静默替换。 */
+ public static void validate(CollectorState state)
+ {
+ validateSampleInterval(state.sampleIntervalMs());
+ if (state.nextSequence() < 1 || state.channels().size() != state.channelCount())
+ throw new IllegalStateException("Invalid collector sampling state in Redis");
+ for (int index = 0; index < state.channels().size(); index++)
+ {
+ CollectorChannelState channel = state.channels().get(index);
+ if (!channelCode(index + 1).equals(channel.channelCode()))
+ throw new IllegalStateException("Invalid collector channel identity in Redis");
+ validateMetrics(channel.metrics());
+ }
+ validateFaults(state.historyFaults());
+ }
+
+ private static List<CollectorMeasurementRecord> faultedHistory(CollectorState current)
+ {
+ CollectorHistoryFaults faults = current.historyFaults();
+ Set<Long> missing = Set.copyOf(faults.missingSequences());
+ List<CollectorMeasurementRecord> records = new ArrayList<>();
+ for (CollectorMeasurementRecord record : current.history())
+ {
+ if (missing.contains(record.sequenceId())) continue;
+ records.add(record);
+ if (Long.valueOf(record.sequenceId()).equals(faults.duplicateSequence())) records.add(record);
+ if (Long.valueOf(record.sequenceId()).equals(faults.conflictingSequence()))
+ records.add(new CollectorMeasurementRecord(record.adapterCode(), record.collectorCode(),
+ record.channelCode(), record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(),
+ record.serverReceivedAt(), record.metricCode(),
+ record.value().add(new BigDecimal("0.001")), record.unit()));
+ }
+ if (faults.outOfOrder())
+ records.sort(Comparator.comparingLong(CollectorMeasurementRecord::sequenceId).reversed()
+ .thenComparing(CollectorMeasurementRecord::channelCode)
+ .thenComparing(CollectorMeasurementRecord::metricCode));
+ return records;
+ }
+
+ private static CollectorState copy(CollectorState current, String ip, int channelCount, boolean online,
+ int sampleIntervalMs, long nextSequence, Instant lastSampledAt,
+ List<CollectorChannelState> channels, List<CollectorMeasurementRecord> history,
+ CollectorHistoryFaults faults)
+ {
+ return new CollectorState(current.collectorId(), current.deviceCode(), ip.trim(), current.mac(),
+ current.name(), current.manufacturer(), current.model(), channelCount, online, sampleIntervalMs,
+ nextSequence, lastSampledAt, channels, history, faults);
+ }
+
+ private static void validateSampleInterval(int sampleIntervalMs)
+ {
+ if (sampleIntervalMs < MIN_SAMPLE_INTERVAL_MS || sampleIntervalMs > MAX_SAMPLE_INTERVAL_MS)
+ throw new IllegalArgumentException("sampleIntervalMs must be between " + MIN_SAMPLE_INTERVAL_MS
+ + " and " + MAX_SAMPLE_INTERVAL_MS);
+ }
+
+ private static void validateMetrics(List<CollectorMetricState> metrics)
+ {
+ if (metrics == null || metrics.isEmpty())
+ throw new IllegalArgumentException("Collector channel must contain at least one metric");
+ java.util.HashSet<String> codes = new java.util.HashSet<>();
+ for (CollectorMetricState metric : metrics)
+ {
+ if (metric.metricCode() == null || metric.metricCode().isBlank() || metric.value() == null
+ || metric.unit() == null || metric.unit().isBlank() || !codes.add(metric.metricCode().trim()))
+ throw new IllegalArgumentException("Collector metric code, value and unit must be valid and unique");
+ }
+ }
+
+ private static void validateFaults(CollectorHistoryFaults faults)
+ {
+ if (faults == null) throw new IllegalArgumentException("historyFaults must not be null");
+ if (faults.missingSequences().stream().anyMatch(sequence -> sequence == null || sequence < 1)
+ || faults.duplicateSequence() != null && faults.duplicateSequence() < 1
+ || faults.conflictingSequence() != null && faults.conflictingSequence() < 1)
+ throw new IllegalArgumentException("History fault sequences must be positive");
+ }
+
+ private static int parseChannelIndex(String value, int channelCount)
+ {
+ if (value == null || !value.matches("S-\\d{2}"))
+ throw new IllegalArgumentException("Collector channel must use S-01...S-N identity");
+ int index = Integer.parseInt(value.substring(2)) - 1;
+ if (index < 0 || index >= channelCount || !channelCode(index + 1).equals(value))
+ throw new IllegalArgumentException("Unknown collector channel: " + value);
+ return index;
+ }
+
+ private static String channelCode(int index)
+ {
+ return String.format("S-%02d", index);
+ }
+
+ public record CollectorRealtimeSnapshot(String collectorCode, boolean online, int sampleIntervalMs,
+ List<CollectorMeasurementRecord> measurements)
+ {
+ public CollectorRealtimeSnapshot { measurements = List.copyOf(measurements); }
+ }
+
+ public record CollectorHistoryPage(String collectorCode, long offset, long nextOffset, boolean hasMore,
+ long total, List<CollectorMeasurementRecord> records)
+ {
+ public CollectorHistoryPage { records = List.copyOf(records); }
+ }
+}
diff --git a/ruoyi-device/src/test/java/com/ruoyi/device/contract/SimulatorPublicContractTest.java b/ruoyi-device/src/test/java/com/ruoyi/device/contract/SimulatorPublicContractTest.java
index de1a699..9e8be45 100644
--- a/ruoyi-device/src/test/java/com/ruoyi/device/contract/SimulatorPublicContractTest.java
+++ b/ruoyi-device/src/test/java/com/ruoyi/device/contract/SimulatorPublicContractTest.java
@@ -1,9 +1,14 @@
package com.ruoyi.device.contract;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import java.math.BigDecimal;
import java.time.Duration;
+import java.time.Instant;
import java.time.ZoneId;
+import java.util.List;
import java.util.Set;
import org.junit.jupiter.api.Test;
import com.ruoyi.device.api.DeviceAdapter;
@@ -16,6 +21,8 @@
import com.ruoyi.device.model.DeviceWorkStatus;
import com.ruoyi.device.simulator.BtsSimulatorStateStore;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.DeviceState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;
import com.ruoyi.device.simulator.InMemoryBtsSimulatorStateStore;
@@ -23,6 +30,89 @@
class SimulatorPublicContractTest
{
@Test
+ void collectorGeneratesIndependentRealtimeAndHistoricalMeasurements() throws Exception
+ {
+ Instant startedAt = Instant.parse("2026-08-29T01:00:00Z");
+ try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0))
+ {
+ simulator.configureCollectorSampling(1, 200, CollectorHistoryFaults.none());
+ simulator.configureCollectorChannel(1, "S-01", List.of(
+ new CollectorMetricState("temperature", new BigDecimal("26.5"), "℃"),
+ new CollectorMetricState("positive_potential", new BigDecimal("3.801"), "V"),
+ new CollectorMetricState("negative_potential", new BigDecimal("-0.120"), "V"),
+ new CollectorMetricState("pressure", new BigDecimal("101.2"), "kPa")));
+
+ simulator.advanceCollectors(startedAt);
+ simulator.advanceCollectors(startedAt.plusMillis(400));
+
+ var realtime = simulator.snapshot().collectors().get(0).latestMeasurements();
+ var history = simulator.collectorHistory(1, 0, 1000);
+ assertEquals(13, realtime.size());
+ assertTrue(realtime.stream().allMatch(item -> item.sequenceId() == 3));
+ assertTrue(realtime.stream().anyMatch(item -> item.channelCode().equals("S-01")
+ && item.metricCode().equals("pressure") && item.unit().equals("kPa")));
+ assertEquals(39, history.total());
+ assertEquals("dev-sim", history.records().get(0).adapterCode());
+ assertEquals("dev-sim-collector-1", history.records().get(0).collectorCode());
+ assertEquals(startedAt, history.records().get(0).deviceTime());
+ assertTrue(history.records().get(0).deviceTimeRaw().startsWith("2026-08-29 09:00:00"));
+ assertEquals(startedAt, history.records().get(0).serverReceivedAt());
+ }
+ }
+
+ @Test
+ void collectorHistorySupportsPagingAndRepeatableFaultScenarios() throws Exception
+ {
+ Instant startedAt = Instant.parse("2026-08-29T02:00:00Z");
+ try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0))
+ {
+ simulator.configureCollectorSampling(1, 100,
+ new CollectorHistoryFaults(List.of(2L), 1L, 3L, true));
+ simulator.advanceCollectors(startedAt);
+ simulator.advanceCollectors(startedAt.plusMillis(200));
+
+ var first = simulator.collectorHistory(1, 0, 17);
+ var second = simulator.collectorHistory(1, first.nextOffset(), 1000);
+ List<com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord> records =
+ new java.util.ArrayList<>(first.records());
+ records.addAll(second.records());
+
+ assertEquals(48, first.total());
+ assertTrue(first.hasMore());
+ assertFalse(second.hasMore());
+ assertTrue(records.stream().noneMatch(item -> item.sequenceId() == 2));
+ assertEquals(24, records.stream().filter(item -> item.sequenceId() == 1).count());
+ assertEquals(2, records.stream().filter(item -> item.sequenceId() == 3
+ && item.channelCode().equals("S-01") && item.metricCode().equals("temperature")).count());
+ assertTrue(records.get(0).sequenceId() >= records.get(records.size() - 1).sequenceId());
+ }
+ }
+
+ @Test
+ void collectorStopsWhileOfflineAndResumesFromPersistedSequence() throws Exception
+ {
+ InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore();
+ Instant now = Instant.now();
+ try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0, store))
+ {
+ simulator.advanceCollectors(now);
+ simulator.updateCollector(1, "127.0.1.2", 4, false);
+ simulator.advanceCollectors(now.plusSeconds(10));
+ assertEquals(1, simulator.snapshot().collectors().get(0).latestMeasurements().get(0).sequenceId());
+ }
+
+ try (BtsDevelopmentSimulator restored = new BtsDevelopmentSimulator(0, store))
+ {
+ restored.updateCollector(1, "127.0.1.2", 4, true);
+ restored.advanceCollectors(Instant.now().plusSeconds(2));
+ assertTrue(restored.snapshot().collectors().get(0).latestMeasurements().get(0).sequenceId() >= 2);
+ assertEquals(List.of("S-01", "S-02", "S-03", "S-04"),
+ restored.snapshot().collectors().get(0).channels().stream()
+ .map(channel -> channel.channelCode()).toList());
+ }
+ }
+
+ @Test
void simulatorIsReusableOutsideTheBtsPackage() throws Exception
{
try (ScriptedBtsTcpSimulator simulator = new ScriptedBtsTcpSimulator();
--
Gitblit v1.7.1