From 370ef4f5e663fdfc1d19d2774dfc4ec1ed0af60e Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 29 八月 2026 14:57:48 +0800
Subject: [PATCH] 修复采集器历史状态持续序列化错误
---
ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java | 45 +++-
ruoyi-device/src/main/java/com/ruoyi/device/simulator/InMemoryBtsSimulatorStateStore.java | 33 +++
ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java | 151 +++++++++++++++-
ruoyi-device/src/main/java/com/ruoyi/device/simulator/BtsSimulatorStateStore.java | 33 +++
ruoyi-admin/src/test/java/com/ruoyi/config/RedisStateStoreTest.java | 122 +++++++++++++
ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java | 56 +++--
ruoyi-device/src/test/java/com/ruoyi/device/contract/SimulatorPublicContractTest.java | 79 +++++++-
7 files changed, 460 insertions(+), 59 deletions(-)
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 0b9bb63..6c9d1eb 100644
--- a/ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java
@@ -4,7 +4,10 @@
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
+import org.springframework.data.redis.core.RedisOperations;
+import org.springframework.data.redis.core.SessionCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson2.JSON;
@@ -19,9 +22,13 @@
{
static final String STATE_KEY = "dev:bts-simulator:state:v1";
static final String RECENT_REQUESTS_KEY = "dev:bts-simulator:recent-requests:v1";
+ static final String HISTORY_KEYS_KEY = "dev:bts-simulator:collector-history-keys:v1";
+ static final String HISTORY_KEY_PREFIX = "dev:bts-simulator:collector-history:v1:";
private static final long RECENT_REQUESTS_TTL_HOURS = 1;
+ private static final int HISTORY_READ_BATCH_SIZE = 1000;
private final RedisTemplate<Object, Object> redisTemplate;
+ private List<CollectorHistoryAppend> pendingEmbeddedHistory = List.of();
public RedisBtsSimulatorStateStore(RedisTemplate<Object, Object> redisTemplate)
{
@@ -32,18 +39,106 @@
public SimulatorState loadOrInitialize(SimulatorState defaultState)
{
Object stored = redisTemplate.opsForValue().get(STATE_KEY);
- if (stored != null) return parseState(stored);
+ if (stored != null) return restoreState(stored);
String initial = JSON.toJSONString(defaultState);
if (Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(STATE_KEY, initial))) return defaultState;
stored = redisTemplate.opsForValue().get(STATE_KEY);
if (stored == null) throw new IllegalStateException("BTS simulator state initialization lost in Redis");
- return parseState(stored);
+ return restoreState(stored);
+ }
+
+ private SimulatorState restoreState(Object stored)
+ {
+ ParsedSimulatorState parsed = parseState(stored);
+ // 旧格式历史先只供恢复校验读取，损坏状态不得被迁移写回覆盖。
+ pendingEmbeddedHistory = parsed.embeddedHistory();
+ return parsed.state();
+ }
+
+ @Override
+ public void completeValidatedRestore(SimulatorState state)
+ {
+ if (pendingEmbeddedHistory.isEmpty()) return;
+ save(state, pendingEmbeddedHistory);
+ pendingEmbeddedHistory = List.of();
}
@Override
public void save(SimulatorState state)
{
- redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(state));
+ save(state, List.of());
+ }
+
+ @Override
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ public void save(SimulatorState state, List<CollectorHistoryAppend> historyAppends)
+ {
+ List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>()
+ {
+ @Override
+ public <K, V> List<Object> execute(RedisOperations<K, V> operations)
+ {
+ RedisOperations<Object, Object> typed = (RedisOperations) operations;
+ typed.multi();
+ for (CollectorHistoryAppend append : historyAppends)
+ {
+ String historyKey = historyKey(append.collectorCode());
+ if (!append.records().isEmpty())
+ {
+ List<Object> records = new ArrayList<>(
+ append.records().stream().map(JSON::toJSONString).toList());
+ typed.opsForList().rightPushAll(historyKey, records);
+ typed.opsForSet().add(HISTORY_KEYS_KEY, historyKey);
+ }
+ }
+ typed.opsForValue().set(STATE_KEY, JSON.toJSONString(state));
+ return typed.exec();
+ }
+ });
+ if (results == null) throw new IllegalStateException("BTS simulator state transaction failed in Redis");
+ }
+
+ @Override
+ public List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
+ {
+ String key = historyKey(collectorCode);
+ Long storedSize = redisTemplate.opsForList().size(key);
+ List<CollectorMeasurementRecord> records = new ArrayList<>();
+ if (storedSize != null)
+ {
+ for (long start = 0; start < storedSize; start += HISTORY_READ_BATCH_SIZE)
+ {
+ long end = Math.min(start + HISTORY_READ_BATCH_SIZE, storedSize) - 1;
+ List<Object> stored = redisTemplate.opsForList().range(key, start, end);
+ if (stored != null)
+ stored.forEach(item -> records.add(collectorMeasurementRecord(item)));
+ }
+ }
+ pendingEmbeddedHistory.stream()
+ .filter(append -> append.collectorCode().equals(collectorCode))
+ .forEach(append -> records.addAll(append.records()));
+ return List.copyOf(records);
+ }
+
+ @Override
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ public void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode)
+ {
+ String historyKey = historyKey(collectorCode);
+ List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>()
+ {
+ @Override
+ public <K, V> List<Object> execute(RedisOperations<K, V> operations)
+ {
+ RedisOperations<Object, Object> typed = (RedisOperations) operations;
+ typed.multi();
+ typed.delete(historyKey);
+ typed.opsForSet().remove(HISTORY_KEYS_KEY, historyKey);
+ typed.opsForValue().set(STATE_KEY, JSON.toJSONString(state));
+ return typed.exec();
+ }
+ });
+ if (results == null) throw new IllegalStateException("BTS simulator collector deletion failed in Redis");
}
@Override
@@ -64,11 +159,15 @@
@Override
public void reset(SimulatorState defaultState)
{
+ pendingEmbeddedHistory = List.of();
+ Set<Object> historyKeys = redisTemplate.opsForSet().members(HISTORY_KEYS_KEY);
+ if (historyKeys != null && !historyKeys.isEmpty()) redisTemplate.delete(historyKeys);
+ redisTemplate.delete(HISTORY_KEYS_KEY);
redisTemplate.delete(RECENT_REQUESTS_KEY);
save(defaultState);
}
- private SimulatorState parseState(Object stored)
+ private ParsedSimulatorState parseState(Object stored)
{
try
{
@@ -80,11 +179,20 @@
List<DeviceState> devices = new ArrayList<>();
requiredArray(root, "devices").forEach(item -> devices.add(deviceState(item)));
List<CollectorState> collectors = new ArrayList<>();
- requiredArray(root, "collectors").forEach(item -> collectors.add(collectorState(item)));
+ List<CollectorHistoryAppend> embeddedHistory = new ArrayList<>();
+ requiredArray(root, "collectors").forEach(item -> {
+ JSONObject collectorObject = jsonObject(item);
+ List<CollectorMeasurementRecord> history = embeddedCollectorHistory(collectorObject);
+ CollectorState collector = collectorState(collectorObject, history);
+ collectors.add(collector);
+ if (!history.isEmpty())
+ embeddedHistory.add(new CollectorHistoryAppend(collector.deviceCode(), history));
+ });
List<ChannelTombstone> tombstones = new ArrayList<>();
requiredArray(root, "tombstones").forEach(item -> tombstones.add(channelTombstone(item)));
- return new SimulatorState(version, devices, collectors,
+ SimulatorState state = new SimulatorState(version, devices, collectors,
requiredInteger(root, "nextDeviceId"), requiredInteger(root, "nextCollectorId"), tombstones);
+ return new ParsedSimulatorState(state, embeddedHistory);
}
catch (IllegalStateException exception)
{
@@ -132,9 +240,17 @@
object.getBooleanValue("online"), channels);
}
- private CollectorState collectorState(Object value)
+ private List<CollectorMeasurementRecord> embeddedCollectorHistory(JSONObject object)
{
- JSONObject object = jsonObject(value);
+ JSONArray values = object.getJSONArray("history");
+ if (values == null) return List.of();
+ List<CollectorMeasurementRecord> history = new ArrayList<>(values.size());
+ values.forEach(item -> history.add(collectorMeasurementRecord(item)));
+ return List.copyOf(history);
+ }
+
+ private CollectorState collectorState(JSONObject object, List<CollectorMeasurementRecord> embeddedHistory)
+ {
int collectorId = requiredInteger(object, "collectorId");
String deviceCode = object.getString("deviceCode");
String ip = object.getString("ip");
@@ -151,18 +267,19 @@
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)));
List<CollectorRealtimeMeasurement> latestMeasurements = new ArrayList<>();
JSONArray latest = object.getJSONArray("latestMeasurements");
if (latest != null)
latest.forEach(item -> latestMeasurements.add(collectorRealtimeMeasurement(item)));
else
- latestMeasurements.addAll(latestMeasurementsFromLegacyHistory(history, channels));
+ latestMeasurements.addAll(latestMeasurementsFromLegacyHistory(embeddedHistory, channels));
+ long historyRecordCount = object.containsKey("historyRecordCount")
+ ? requiredLong(object, "historyRecordCount") : embeddedHistory.size();
return new CollectorState(collectorId, deviceCode, ip, mac, name, manufacturer, model,
channelCount, online, requiredInteger(object, "sampleIntervalMs"),
requiredLong(object, "nextSequence"), optionalInstant(object, "lastSampledAt"),
- channels, latestMeasurements, history, collectorHistoryFaults(object.get("historyFaults")));
+ channels, latestMeasurements, historyRecordCount,
+ collectorHistoryFaults(object.get("historyFaults")));
}
private CollectorChannelState collectorChannelState(Object value)
@@ -270,4 +387,14 @@
String value = object.getString(field);
return value == null ? null : Instant.parse(value);
}
+
+ private String historyKey(String collectorCode)
+ {
+ return HISTORY_KEY_PREFIX + collectorCode;
+ }
+
+ private record ParsedSimulatorState(SimulatorState state, List<CollectorHistoryAppend> embeddedHistory)
+ {
+ private ParsedSimulatorState { embeddedHistory = List.copyOf(embeddedHistory); }
+ }
}
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 6ccc48d..d9fad0c 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/config/RedisStateStoreTest.java
+++ b/ruoyi-admin/src/test/java/com/ruoyi/config/RedisStateStoreTest.java
@@ -3,17 +3,26 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.Test;
import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ListOperations;
+import org.springframework.data.redis.core.SessionCallback;
+import org.springframework.data.redis.core.SetOperations;
import org.springframework.data.redis.core.ValueOperations;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.device.model.DeviceWorkStatus;
+import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelTombstone;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState;
@@ -78,7 +87,65 @@
assertEquals(1L, collector.nextSequence());
assertEquals(List.of("S-01", "S-02"),
collector.channels().stream().map(channel -> channel.channelCode()).toList());
- assertEquals(List.of(), collector.history());
+ assertEquals(0L, collector.historyRecordCount());
+ }
+
+ @Test
+ void migratesEmbeddedCollectorHistoryToAppendOnlyRedisList()
+ {
+ 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\":1,\"online\":true,"
+ + "\"sampleIntervalMs\":1000,\"nextSequence\":2,"
+ + "\"lastSampledAt\":\"2026-08-29T05:00:00Z\","
+ + "\"channels\":[{\"channelCode\":\"S-01\",\"metrics\":[{"
+ + "\"metricCode\":\"temperature\",\"value\":25.1,\"unit\":\"℃\"}]}],"
+ + "\"history\":[{\"adapterCode\":\"dev-sim\","
+ + "\"collectorCode\":\"dev-sim-collector-1\",\"channelCode\":\"S-01\","
+ + "\"sequenceId\":1,\"deviceTime\":\"2026-08-29T05:00:00Z\","
+ + "\"deviceTimeRaw\":\"2026-08-29 13:00:00.000\","
+ + "\"serverReceivedAt\":\"2026-08-29T05:00:00Z\","
+ + "\"metricCode\":\"temperature\",\"value\":25.1,\"unit\":\"℃\"}],"
+ + "\"historyFaults\":{\"missingSequences\":[],\"outOfOrder\":false}}],"
+ + "\"nextDeviceId\":1,\"nextCollectorId\":2,\"tombstones\":[]}");
+ RedisBtsSimulatorStateStore store = store(stored);
+
+ SimulatorState restored = store.loadOrInitialize(state(1, "127.0.0.2"));
+
+ assertEquals(1L, restored.collectors().get(0).historyRecordCount());
+ assertEquals(1, store.loadCollectorHistory("dev-sim-collector-1").size());
+ assertEquals(true, stored.get().toString().contains("\"history\":[{"));
+ store.completeValidatedRestore(restored);
+ assertEquals(false, stored.get().toString().contains("\"history\":[{"));
+ }
+
+ @Test
+ void doesNotOverwriteDamagedEmbeddedHistoryBeforeValidation()
+ {
+ 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\":1,\"online\":true,"
+ + "\"sampleIntervalMs\":1000,\"nextSequence\":2,"
+ + "\"lastSampledAt\":\"2026-08-29T05:00:00Z\","
+ + "\"channels\":[{\"channelCode\":\"S-01\",\"metrics\":[{"
+ + "\"metricCode\":\"temperature\",\"value\":25.1,\"unit\":\"℃\"}]}],"
+ + "\"history\":[{\"adapterCode\":\"dev-sim\","
+ + "\"collectorCode\":\"wrong-collector\",\"channelCode\":\"S-01\","
+ + "\"sequenceId\":1,\"deviceTime\":\"2026-08-29T05:00:00Z\","
+ + "\"deviceTimeRaw\":\"2026-08-29 13:00:00.000\","
+ + "\"serverReceivedAt\":\"2026-08-29T05:00:00Z\","
+ + "\"metricCode\":\"temperature\",\"value\":25.1,\"unit\":\"℃\"}],"
+ + "\"historyFaults\":{\"missingSequences\":[],\"outOfOrder\":false}}],"
+ + "\"nextDeviceId\":1,\"nextCollectorId\":2,\"tombstones\":[]}");
+
+ assertThrows(IllegalStateException.class,
+ () -> new BtsDevelopmentSimulator(0, store(stored)));
+
+ assertEquals(true, stored.get().toString().contains("\"history\":[{"));
}
@Test
@@ -93,6 +160,29 @@
assertEquals("Unsupported BTS simulator state version: 2", exception.getMessage());
assertEquals(true, stored.get().toString().contains("\"version\":2"));
+ }
+
+ @Test
+ void atomicallySavesCollectorDeletionAndRemovesIndependentHistory()
+ {
+ AtomicReference<Object> stored = new AtomicReference<>();
+ RedisBtsSimulatorStateStore store = store(stored);
+ SimulatorState current = state(1, "127.0.0.2");
+ store.loadOrInitialize(current);
+ var record = new com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord(
+ "dev-sim", "dev-sim-collector-1", "S-01", 1L,
+ java.time.Instant.parse("2026-08-29T05:00:00Z"), "2026-08-29 13:00:00.000",
+ java.time.Instant.parse("2026-08-29T05:00:00Z"), "temperature",
+ new java.math.BigDecimal("25.1"), "℃");
+ store.save(current, List.of(
+ new com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryAppend(
+ "dev-sim-collector-1", List.of(record))));
+ SimulatorState next = state(7, "127.0.0.9");
+
+ store.saveAndDeleteCollectorHistory(next, "dev-sim-collector-1");
+
+ assertEquals(next, store.loadOrInitialize(current));
+ assertEquals(List.of(), store.loadCollectorHistory("dev-sim-collector-1"));
}
@Test
@@ -175,7 +265,37 @@
{
RedisTemplate<Object, Object> redis = mock(RedisTemplate.class);
ValueOperations<Object, Object> values = mock(ValueOperations.class);
+ ListOperations<Object, Object> lists = mock(ListOperations.class);
+ SetOperations<Object, Object> sets = mock(SetOperations.class);
+ Map<Object, List<Object>> listValues = new LinkedHashMap<>();
when(redis.opsForValue()).thenReturn(values);
+ when(redis.opsForList()).thenReturn(lists);
+ when(redis.opsForSet()).thenReturn(sets);
+ when(sets.add(any(), any())).thenReturn(1L);
+ when(sets.members(any())).thenReturn(Set.of());
+ when(redis.exec()).thenReturn(List.of(Boolean.TRUE));
+ when(redis.execute(any(SessionCallback.class))).thenAnswer(invocation -> {
+ SessionCallback<?> callback = invocation.getArgument(0);
+ return callback.execute(redis);
+ });
+ when(lists.rightPushAll(any(), org.mockito.ArgumentMatchers.<Object>anyCollection()))
+ .thenAnswer(invocation -> {
+ Object key = invocation.getArgument(0);
+ List<Object> records = new ArrayList<>(listValues.getOrDefault(key, List.of()));
+ FastJson2JsonRedisSerializer<Object> serializer =
+ new FastJson2JsonRedisSerializer<>(Object.class);
+ List<Object> appended = invocation.getArgument(1);
+ appended.forEach(item -> records.add(
+ serializer.deserialize(serializer.serialize(item))));
+ listValues.put(key, records);
+ return (long) records.size();
+ });
+ when(lists.range(any(), anyLong(), anyLong())).thenAnswer(invocation ->
+ List.copyOf(listValues.getOrDefault(invocation.getArgument(0), List.of())));
+ when(lists.size(any())).thenAnswer(invocation ->
+ (long) listValues.getOrDefault(invocation.getArgument(0), List.of()).size());
+ when(redis.delete(org.mockito.ArgumentMatchers.<Object>any()))
+ .thenAnswer(invocation -> listValues.remove(invocation.getArgument(0)) != null);
when(values.get(any())).thenAnswer(invocation -> stored.get());
when(values.setIfAbsent(any(), any())).thenAnswer(invocation -> {
if (stored.get() != null) return false;
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 1663a40..5548804 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
@@ -13,6 +13,7 @@
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.CollectorHistoryAppend;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
@@ -23,6 +24,7 @@
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.CollectorAdvance;
import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorHistoryPage;
import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
@@ -58,6 +60,7 @@
this.stateStore = stateStore;
state = stateStore.loadOrInitialize(defaultState());
validateRestoredState();
+ stateStore.completeValidatedRestore(state);
simulator = ScriptedBtsTcpSimulator.responding(port, this::respond);
}
@@ -155,11 +158,13 @@
/** 删除指定开发采集器，不影响检测柜 TCP 发现结果。 */
public synchronized void deleteCollector(int collectorId)
{
- requireCollector(collectorId);
+ CollectorState collector = requireCollector(collectorId);
List<CollectorState> collectors = new ArrayList<>(state.collectors());
collectors.removeIf(item -> item.collectorId() == collectorId);
- persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
- state.nextCollectorId(), state.tombstones()));
+ SimulatorState next = new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
+ state.nextCollectorId(), state.tombstones());
+ stateStore.saveAndDeleteCollectorHistory(next, collector.deviceCode());
+ this.state = next;
}
/** 配置采集器独立采样频率和历史下载故障场景。 */
@@ -181,12 +186,18 @@
/** 按每台采集器自己的频率推进实时和历史采样。 */
public synchronized void advanceCollectors(Instant now)
{
- List<CollectorState> collectors = state.collectors().stream()
- .map(collector -> DevelopmentCollectorSimulation.advance(collector, now))
- .toList();
- if (!collectors.equals(state.collectors()))
+ List<CollectorState> collectors = new ArrayList<>(state.collectors().size());
+ List<CollectorHistoryAppend> historyAppends = new ArrayList<>();
+ for (CollectorState collector : state.collectors())
+ {
+ CollectorAdvance advance = DevelopmentCollectorSimulation.advance(collector, now);
+ collectors.add(advance.state());
+ if (!advance.appendedRecords().isEmpty())
+ historyAppends.add(new CollectorHistoryAppend(collector.deviceCode(), advance.appendedRecords()));
+ }
+ if (!historyAppends.isEmpty())
persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
- state.nextCollectorId(), state.tombstones()));
+ state.nextCollectorId(), state.tombstones()), historyAppends);
}
/** 返回采集器最新实时快照；快照与分页历史保持不同读取路径。 */
@@ -198,7 +209,9 @@
/** 分页下载采集器模拟历史，并应用已配置的重复、冲突、乱序或缺失场景。 */
public synchronized CollectorHistoryPage collectorHistory(int collectorId, long offset, int limit)
{
- return DevelopmentCollectorSimulation.history(requireCollector(collectorId), offset, limit);
+ CollectorState collector = requireCollector(collectorId);
+ return DevelopmentCollectorSimulation.history(collector,
+ stateStore.loadCollectorHistory(collector.deviceCode()), offset, limit);
}
/** 返回控制页需要的运行状态快照。 */
@@ -215,7 +228,7 @@
.map(collector -> new CollectorSnapshot(collector.collectorId(), collector.deviceCode(),
collector.ip(), collector.mac(), collector.name(), collector.manufacturer(),
collector.model(), collector.channelCount(), collector.online(), collector.sampleIntervalMs(),
- collector.channels(), collector.historyFaults(), collector.history().size(),
+ collector.channels(), collector.historyFaults(), collector.historyRecordCount(),
DevelopmentCollectorSimulation.realtime(collector).measurements()))
.toList();
return new StateSnapshot(running, HOST + ":" + port(), deviceSnapshots, collectorSnapshots,
@@ -452,6 +465,13 @@
state = nextState;
}
+ private void persist(SimulatorState nextState, List<CollectorHistoryAppend> historyAppends)
+ {
+ // 主状态与原始历史批次必须一起成功，内存序号才能向前推进。
+ stateStore.save(nextState, historyAppends);
+ state = nextState;
+ }
+
private String simulatorMac(int devtype, int devid)
{
return String.format("02:44:%02X:%02X:%02X:%02X",
@@ -498,7 +518,8 @@
|| !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);
+ DevelopmentCollectorSimulation.validate(collector,
+ stateStore.loadCollectorHistory(collector.deviceCode()));
maximumCollectorId = Math.max(maximumCollectorId, collector.collectorId());
}
java.util.Set<String> tombstoneIds = new java.util.HashSet<>();
@@ -529,7 +550,7 @@
public record CollectorSnapshot(int collectorId, String deviceCode, String ip, String mac, String name,
String manufacturer, String model, int channelCount, boolean online, int sampleIntervalMs,
- List<CollectorChannelState> channels, CollectorHistoryFaults historyFaults, int historyRecordCount,
+ List<CollectorChannelState> channels, CollectorHistoryFaults historyFaults, long historyRecordCount,
List<CollectorRealtimeMeasurement> latestMeasurements)
{
public CollectorSnapshot
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 7f70ae5..e417b5a 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
@@ -16,6 +16,29 @@
/** 完整保存下一版模拟器状态。 */
void save(SimulatorState state);
+ /** 状态和独立历史全部校验通过后，提交兼容旧格式所需的迁移。 */
+ default void completeValidatedRestore(SimulatorState state) { }
+
+ /** 原子保存主状态并追加本轮新产生的原始历史。 */
+ default void save(SimulatorState state, List<CollectorHistoryAppend> historyAppends)
+ {
+ if (!historyAppends.isEmpty())
+ throw new UnsupportedOperationException("Collector history append is not supported");
+ save(state);
+ }
+
+ /** 读取指定采集器的只追加原始历史。 */
+ default List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
+ {
+ return List.of();
+ }
+
+ /** 原子保存删除采集器后的主状态，并删除该采集器的独立历史。 */
+ default void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode)
+ {
+ throw new UnsupportedOperationException("Atomic collector deletion is not supported");
+ }
+
/** 读取短期保留的最近请求。 */
List<RecentRequest> loadRecentRequests();
@@ -47,13 +70,12 @@
String manufacturer, String model, int channelCount, boolean online, int sampleIntervalMs,
long nextSequence, Instant lastSampledAt, List<CollectorChannelState> channels,
List<CollectorRealtimeMeasurement> latestMeasurements,
- List<CollectorMeasurementRecord> history, CollectorHistoryFaults historyFaults)
+ long historyRecordCount, CollectorHistoryFaults historyFaults)
{
public CollectorState
{
channels = List.copyOf(channels);
latestMeasurements = List.copyOf(latestMeasurements);
- history = List.copyOf(history);
historyFaults = historyFaults == null ? CollectorHistoryFaults.none() : historyFaults;
}
@@ -63,7 +85,7 @@
{
this(collectorId, deviceCode, ip, mac, name, manufacturer, model, channelCount, online,
DevelopmentCollectorSimulation.DEFAULT_SAMPLE_INTERVAL_MS, 1L, null,
- DevelopmentCollectorSimulation.defaultChannels(channelCount), List.of(), List.of(),
+ DevelopmentCollectorSimulation.defaultChannels(channelCount), List.of(), 0L,
CollectorHistoryFaults.none());
}
}
@@ -83,6 +105,11 @@
long sequenceId, Instant deviceTime, String deviceTimeRaw, Instant serverReceivedAt,
String metricCode, BigDecimal value, String unit) { }
+ record CollectorHistoryAppend(String collectorCode, List<CollectorMeasurementRecord> records)
+ {
+ public CollectorHistoryAppend { records = List.copyOf(records); }
+ }
+
record CollectorHistoryFaults(List<Long> missingSequences, Long duplicateSequence,
Long conflictingSequence, boolean outOfOrder)
{
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
index 369172a..c07d403 100644
--- a/ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java
+++ b/ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java
@@ -61,7 +61,7 @@
List<CollectorRealtimeMeasurement> latestMeasurements = retainConfiguredMeasurements(
current.latestMeasurements(), channels);
return copy(current, ip, channelCount, online, current.sampleIntervalMs(), current.nextSequence(),
- lastSampledAt, channels, latestMeasurements, current.history(), current.historyFaults());
+ lastSampledAt, channels, latestMeasurements, current.historyRecordCount(), current.historyFaults());
}
/** 更新采样频率和历史下载故障；测量原始历史保持只追加。 */
@@ -72,7 +72,7 @@
validateFaults(historyFaults);
return copy(current, current.ip(), current.channelCount(), current.online(), sampleIntervalMs,
current.nextSequence(), current.lastSampledAt(), current.channels(), current.latestMeasurements(),
- current.history(), historyFaults);
+ current.historyRecordCount(), historyFaults);
}
/** 替换一个演示通道的指标配置；动态添加项与固定指标共用指标编码合同。 */
@@ -87,19 +87,19 @@
current.latestMeasurements(), channels);
return copy(current, current.ip(), current.channelCount(), current.online(), current.sampleIntervalMs(),
current.nextSequence(), current.lastSampledAt(), channels, latestMeasurements,
- current.history(), current.historyFaults());
+ current.historyRecordCount(), current.historyFaults());
}
/** 按采集器自身频率推进采样；离线期间不生成记录，恢复后从下一序号继续。 */
- public static CollectorState advance(CollectorState current, Instant now)
+ public static CollectorAdvance advance(CollectorState current, Instant now)
{
- if (!current.online()) return current;
+ if (!current.online()) return new CollectorAdvance(current, List.of());
Instant sampleAt = current.lastSampledAt() == null
? now
: current.lastSampledAt().plusMillis(current.sampleIntervalMs());
- if (sampleAt.isAfter(now)) return current;
+ if (sampleAt.isAfter(now)) return new CollectorAdvance(current, List.of());
- List<CollectorMeasurementRecord> history = new ArrayList<>(current.history());
+ List<CollectorMeasurementRecord> appendedRecords = new ArrayList<>();
Map<String, CollectorRealtimeMeasurement> latest = new LinkedHashMap<>();
for (CollectorRealtimeMeasurement measurement : current.latestMeasurements())
latest.put(measurementKey(measurement.channelCode(), measurement.metricCode()), measurement);
@@ -116,7 +116,7 @@
channel.channelCode(), sequence, sampleAt,
DEVICE_TIME_FORMATTER.format(sampleAt.atZone(DEVICE_ZONE)), now,
metric.metricCode(), metric.value(), metric.unit());
- history.add(record);
+ appendedRecords.add(record);
latest.put(measurementKey(record.channelCode(), record.metricCode()),
new CollectorRealtimeMeasurement(record.adapterCode(), record.collectorCode(),
record.channelCode(), record.sequenceId(), record.deviceTime(),
@@ -129,9 +129,12 @@
generated++;
sampleAt = sampleAt.plusMillis(current.sampleIntervalMs());
}
- return copy(current, current.ip(), current.channelCount(), current.online(), current.sampleIntervalMs(),
- sequence, lastSampledAt, current.channels(), List.copyOf(latest.values()), history,
+ CollectorState next = copy(current, current.ip(), current.channelCount(), current.online(),
+ current.sampleIntervalMs(),
+ sequence, lastSampledAt, current.channels(), List.copyOf(latest.values()),
+ current.historyRecordCount() + appendedRecords.size(),
current.historyFaults());
+ return new CollectorAdvance(next, appendedRecords);
}
/** 返回独立保存的最新实时快照，不从原始历史反推当前状态。 */
@@ -142,13 +145,14 @@
}
/** 对只追加原始模拟历史应用可重复的下载故障视图，再按偏移量分页。 */
- public static CollectorHistoryPage history(CollectorState current, long offset, int limit)
+ public static CollectorHistoryPage history(CollectorState current, List<CollectorMeasurementRecord> rawHistory,
+ 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);
+ List<CollectorMeasurementRecord> download = faultedHistory(current, rawHistory);
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(),
@@ -156,7 +160,7 @@
}
/** 校验 Redis 恢复的采集器采样状态，拒绝损坏状态被静默替换。 */
- public static void validate(CollectorState state)
+ public static void validate(CollectorState state, List<CollectorMeasurementRecord> rawHistory)
{
validateSampleInterval(state.sampleIntervalMs());
if (state.nextSequence() < 1 || state.channels().size() != state.channelCount())
@@ -169,7 +173,7 @@
validateMetrics(channel.metrics());
}
validateLatestMeasurements(state);
- validateHistory(state);
+ validateHistory(state, rawHistory);
validateFaults(state.historyFaults());
}
@@ -189,12 +193,12 @@
}
}
- private static void validateHistory(CollectorState state)
+ private static void validateHistory(CollectorState state, List<CollectorMeasurementRecord> rawHistory)
{
long maximumSequence = 0;
long previousSequence = 0;
Set<String> identities = new java.util.HashSet<>();
- for (CollectorMeasurementRecord record : state.history())
+ for (CollectorMeasurementRecord record : rawHistory)
{
String identity = record.sequenceId() + "\u0000" + record.channelCode() + "\u0000" + record.metricCode();
if (!validMeasurement(state, record.adapterCode(), record.collectorCode(), record.channelCode(),
@@ -207,6 +211,8 @@
}
if (state.nextSequence() <= maximumSequence)
throw new IllegalStateException("Invalid collector history sequence in Redis");
+ if (state.historyRecordCount() != rawHistory.size())
+ throw new IllegalStateException("Invalid collector history count in Redis");
}
private static boolean validMeasurement(CollectorState state, String adapterCode, String collectorCode,
@@ -266,12 +272,13 @@
return channelCode + "\u0000" + metricCode;
}
- private static List<CollectorMeasurementRecord> faultedHistory(CollectorState current)
+ private static List<CollectorMeasurementRecord> faultedHistory(CollectorState current,
+ List<CollectorMeasurementRecord> rawHistory)
{
CollectorHistoryFaults faults = current.historyFaults();
Set<Long> missing = Set.copyOf(faults.missingSequences());
List<CollectorMeasurementRecord> records = new ArrayList<>();
- for (CollectorMeasurementRecord record : current.history())
+ for (CollectorMeasurementRecord record : rawHistory)
{
if (missing.contains(record.sequenceId())) continue;
records.add(record);
@@ -292,12 +299,12 @@
private static CollectorState copy(CollectorState current, String ip, int channelCount, boolean online,
int sampleIntervalMs, long nextSequence, Instant lastSampledAt,
List<CollectorChannelState> channels, List<CollectorRealtimeMeasurement> latestMeasurements,
- List<CollectorMeasurementRecord> history,
+ long historyRecordCount,
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, latestMeasurements, history, faults);
+ nextSequence, lastSampledAt, channels, latestMeasurements, historyRecordCount, faults);
}
private static void validateSampleInterval(int sampleIntervalMs)
@@ -315,7 +322,9 @@
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()))
+ || !metric.metricCode().equals(metric.metricCode().trim())
+ || metric.unit() == null || metric.unit().isBlank()
+ || !metric.unit().equals(metric.unit().trim()) || !codes.add(metric.metricCode()))
throw new IllegalArgumentException("Collector metric code, value and unit must be valid and unique");
}
}
@@ -350,6 +359,11 @@
public CollectorRealtimeSnapshot { measurements = List.copyOf(measurements); }
}
+ public record CollectorAdvance(CollectorState state, List<CollectorMeasurementRecord> appendedRecords)
+ {
+ public CollectorAdvance { appendedRecords = List.copyOf(appendedRecords); }
+ }
+
public record CollectorHistoryPage(String collectorCode, long offset, long nextOffset, boolean hasMore,
long total, List<CollectorMeasurementRecord> records)
{
diff --git a/ruoyi-device/src/main/java/com/ruoyi/device/simulator/InMemoryBtsSimulatorStateStore.java b/ruoyi-device/src/main/java/com/ruoyi/device/simulator/InMemoryBtsSimulatorStateStore.java
index 780ffe3..1c6db82 100644
--- a/ruoyi-device/src/main/java/com/ruoyi/device/simulator/InMemoryBtsSimulatorStateStore.java
+++ b/ruoyi-device/src/main/java/com/ruoyi/device/simulator/InMemoryBtsSimulatorStateStore.java
@@ -1,6 +1,11 @@
package com.ruoyi.device.simulator;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryAppend;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.RecentRequest;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;
@@ -9,6 +14,7 @@
{
private SimulatorState state;
private List<RecentRequest> recentRequests = List.of();
+ private final Map<String, List<CollectorMeasurementRecord>> collectorHistory = new LinkedHashMap<>();
public InMemoryBtsSimulatorStateStore() { }
@@ -31,6 +37,32 @@
}
@Override
+ public synchronized void save(SimulatorState state, List<CollectorHistoryAppend> historyAppends)
+ {
+ for (CollectorHistoryAppend append : historyAppends)
+ {
+ List<CollectorMeasurementRecord> records = new ArrayList<>(
+ collectorHistory.getOrDefault(append.collectorCode(), List.of()));
+ records.addAll(append.records());
+ collectorHistory.put(append.collectorCode(), List.copyOf(records));
+ }
+ this.state = state;
+ }
+
+ @Override
+ public synchronized List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
+ {
+ return collectorHistory.getOrDefault(collectorCode, List.of());
+ }
+
+ @Override
+ public synchronized void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode)
+ {
+ collectorHistory.remove(collectorCode);
+ this.state = state;
+ }
+
+ @Override
public synchronized List<RecentRequest> loadRecentRequests()
{
return recentRequests;
@@ -47,5 +79,6 @@
{
state = defaultState;
recentRequests = List.of();
+ collectorHistory.clear();
}
}
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 ba19503..7a55294 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
@@ -11,6 +11,7 @@
import java.util.List;
import java.util.Set;
import org.junit.jupiter.api.Test;
+import com.alibaba.fastjson2.JSON;
import com.ruoyi.device.api.DeviceAdapter;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.bts.BtsDeviceAdapter;
@@ -22,6 +23,7 @@
import com.ruoyi.device.simulator.BtsSimulatorStateStore;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorChannelState;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryAppend;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
@@ -132,6 +134,20 @@
}
@Test
+ void collectorRawHistoryIsNotEmbeddedInMainSimulatorState() throws Exception
+ {
+ InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore();
+ try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0, store))
+ {
+ simulator.advanceCollectors(Instant.parse("2026-08-29T03:30:00Z"));
+
+ String mainState = JSON.toJSONString(store.loadOrInitialize(null));
+ assertFalse(mainState.contains("\"history\":[{"));
+ assertEquals(12, simulator.collectorHistory(1, 0, 100).total());
+ }
+ }
+
+ @Test
void collectorRealtimeDropsMeasurementsRemovedFromCurrentConfiguration() throws Exception
{
Instant startedAt = Instant.parse("2026-08-29T04:00:00Z");
@@ -159,6 +175,18 @@
{
assertEquals(4, restored.snapshot().collectors().get(0).latestMeasurements().size());
assertEquals(12, restored.collectorHistory(1, 0, 100).total());
+ }
+ }
+
+ @Test
+ void collectorRejectsMetricWhitespaceThatWouldBreakRestore() throws Exception
+ {
+ try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0))
+ {
+ assertThrows(IllegalArgumentException.class, () -> simulator.configureCollectorChannel(1, "S-01",
+ List.of(new CollectorMetricState("temperature ", new BigDecimal("25.0"), "℃"))));
+ assertThrows(IllegalArgumentException.class, () -> simulator.configureCollectorChannel(1, "S-01",
+ List.of(new CollectorMetricState("temperature", new BigDecimal("25.0"), " ℃"))));
}
}
@@ -347,6 +375,31 @@
}
@Test
+ void collectorDeletionDoesNotReplaceMemoryWhenAtomicPersistenceFails() throws Exception
+ {
+ InMemoryBtsSimulatorStateStore delegate = new InMemoryBtsSimulatorStateStore();
+ BtsSimulatorStateStore failing = new BtsSimulatorStateStore()
+ {
+ @Override public SimulatorState loadOrInitialize(SimulatorState value)
+ { return delegate.loadOrInitialize(value); }
+ @Override public void save(SimulatorState value) { delegate.save(value); }
+ @Override public void saveAndDeleteCollectorHistory(SimulatorState value, String collectorCode)
+ { throw new IllegalStateException("redis unavailable"); }
+ @Override public java.util.List<RecentRequest> loadRecentRequests()
+ { return delegate.loadRecentRequests(); }
+ @Override public void saveRecentRequests(java.util.List<RecentRequest> requests)
+ { delegate.saveRecentRequests(requests); }
+ @Override public void reset(SimulatorState value) { delegate.reset(value); }
+ };
+
+ try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0, failing))
+ {
+ assertThrows(IllegalStateException.class, () -> simulator.deleteCollector(1));
+ assertEquals(1, simulator.snapshot().collectors().size());
+ }
+ }
+
+ @Test
void developmentSimulatorRestoresUnfinishedTombstones() throws Exception
{
InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore();
@@ -390,17 +443,20 @@
Instant observedAt = Instant.parse("2026-08-29T05:00:00Z");
List<CollectorChannelState> channels = List.of(new CollectorChannelState("S-01", List.of(
new CollectorMetricState("temperature", new BigDecimal("25.0"), "℃"))));
+ CollectorMeasurementRecord record = new CollectorMeasurementRecord(
+ "dev-sim", "wrong-collector", "S-01", 1L, observedAt,
+ "2026-08-29 13:00:00.000", observedAt, "temperature",
+ new BigDecimal("25.0"), "℃");
CollectorState collector = new CollectorState(1, "dev-sim-collector-1", "127.0.1.2",
"02:43:00:00:00:01", "模拟采集器", "开发模拟器", "Virtual Collector", 1, true,
- 1000, 2L, observedAt, channels, List.of(), List.of(new CollectorMeasurementRecord(
- "dev-sim", "wrong-collector", "S-01", 1L, observedAt,
- "2026-08-29 13:00:00.000", observedAt, "temperature",
- new BigDecimal("25.0"), "℃")), CollectorHistoryFaults.none());
+ 1000, 2L, observedAt, channels, List.of(), 1L, CollectorHistoryFaults.none());
SimulatorState damaged = new SimulatorState(BtsSimulatorStateStore.CURRENT_VERSION,
List.of(), List.of(collector), 1, 2, List.of());
+ InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore(damaged);
+ store.save(damaged, List.of(new CollectorHistoryAppend(collector.deviceCode(), List.of(record))));
assertThrows(IllegalStateException.class,
- () -> new BtsDevelopmentSimulator(0, new InMemoryBtsSimulatorStateStore(damaged)));
+ () -> new BtsDevelopmentSimulator(0, store));
}
@Test
@@ -409,16 +465,19 @@
Instant observedAt = Instant.parse("2026-08-29T06:00:00Z");
List<CollectorChannelState> channels = List.of(new CollectorChannelState("S-01", List.of(
new CollectorMetricState("temperature", new BigDecimal("25.0"), "℃"))));
+ CollectorMeasurementRecord record = new CollectorMeasurementRecord(
+ "dev-sim", "dev-sim-collector-1", "S-01", 1L, observedAt,
+ "2026-08-29 14:00:00.000", observedAt, "temperature",
+ new BigDecimal("25.0"), "℃");
CollectorState collector = new CollectorState(1, "dev-sim-collector-1", "127.0.1.2",
"02:43:00:00:00:01", "模拟采集器", "开发模拟器", "Virtual Collector", 1, true,
- 1000, 1L, observedAt, channels, List.of(), List.of(new CollectorMeasurementRecord(
- "dev-sim", "dev-sim-collector-1", "S-01", 1L, observedAt,
- "2026-08-29 14:00:00.000", observedAt, "temperature",
- new BigDecimal("25.0"), "℃")), CollectorHistoryFaults.none());
+ 1000, 1L, observedAt, channels, List.of(), 1L, CollectorHistoryFaults.none());
SimulatorState damaged = new SimulatorState(BtsSimulatorStateStore.CURRENT_VERSION,
List.of(), List.of(collector), 1, 2, List.of());
+ InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore(damaged);
+ store.save(damaged, List.of(new CollectorHistoryAppend(collector.deviceCode(), List.of(record))));
assertThrows(IllegalStateException.class,
- () -> new BtsDevelopmentSimulator(0, new InMemoryBtsSimulatorStateStore(damaged)));
+ () -> new BtsDevelopmentSimulator(0, store));
}
}
--
Gitblit v1.7.1