package com.ruoyi.config; import java.math.BigDecimal; 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; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.ruoyi.device.simulator.BtsSimulatorStateStore; import com.ruoyi.device.model.DeviceWorkStatus; /** 将开发模拟器完整状态写入 Redis,主状态不设置普通 TTL。 */ @Component public class RedisBtsSimulatorStateStore implements BtsSimulatorStateStore { 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 redisTemplate; private List pendingEmbeddedHistory = List.of(); public RedisBtsSimulatorStateStore(RedisTemplate redisTemplate) { this.redisTemplate = redisTemplate; } @Override public SimulatorState loadOrInitialize(SimulatorState defaultState) { Object stored = redisTemplate.opsForValue().get(STATE_KEY); 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 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) { save(state, List.of()); } @Override @SuppressWarnings({ "rawtypes", "unchecked" }) public void save(SimulatorState state, List historyAppends) { List results = redisTemplate.execute(new SessionCallback>() { @Override public List execute(RedisOperations operations) { RedisOperations typed = (RedisOperations) operations; typed.multi(); for (CollectorHistoryAppend append : historyAppends) { String historyKey = historyKey(append.collectorCode()); if (!append.records().isEmpty()) { List 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 loadCollectorHistory(String collectorCode) { String key = historyKey(collectorCode); Long storedSize = redisTemplate.opsForList().size(key); List 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 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 results = redisTemplate.execute(new SessionCallback>() { @Override public List execute(RedisOperations operations) { RedisOperations 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 public List loadRecentRequests() { Object stored = redisTemplate.opsForValue().get(RECENT_REQUESTS_KEY); if (stored == null) return List.of(); return List.copyOf(JSON.parseArray(stored.toString(), RecentRequest.class)); } @Override public void saveRecentRequests(List requests) { redisTemplate.opsForValue().set(RECENT_REQUESTS_KEY, JSON.toJSONString(requests), RECENT_REQUESTS_TTL_HOURS, TimeUnit.HOURS); } @Override public void reset(SimulatorState defaultState) { pendingEmbeddedHistory = List.of(); Set 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 ParsedSimulatorState parseState(Object stored) { try { // 通用 RedisTemplate 在可执行包中可能将泛型数组恢复为 JSONObject,必须在存储边界逐层重建。 JSONObject root = jsonObject(stored); Integer version = root.getInteger("version"); if (version == null || version != CURRENT_VERSION) throw new IllegalStateException("Unsupported BTS simulator state version: " + version); List devices = new ArrayList<>(); requiredArray(root, "devices").forEach(item -> devices.add(deviceState(item))); List collectors = new ArrayList<>(); List embeddedHistory = new ArrayList<>(); requiredArray(root, "collectors").forEach(item -> { JSONObject collectorObject = jsonObject(item); List history = embeddedCollectorHistory(collectorObject); CollectorState collector = collectorState(collectorObject, history); collectors.add(collector); if (!history.isEmpty()) embeddedHistory.add(new CollectorHistoryAppend(collector.deviceCode(), history)); }); List tombstones = new ArrayList<>(); requiredArray(root, "tombstones").forEach(item -> tombstones.add(channelTombstone(item))); SimulatorState state = new SimulatorState(version, devices, collectors, requiredInteger(root, "nextDeviceId"), requiredInteger(root, "nextCollectorId"), tombstones); return new ParsedSimulatorState(state, embeddedHistory); } catch (IllegalStateException exception) { throw exception; } catch (Exception exception) { throw new IllegalStateException("Invalid BTS simulator state in Redis", exception); } } private JSONObject jsonObject(Object value) { if (value instanceof JSONObject object) return object; if (value instanceof String text) { Object parsed = JSON.parse(text); if (parsed instanceof String nested) parsed = JSON.parse(nested); if (parsed instanceof JSONObject object) return object; return JSON.parseObject(JSON.toJSONString(parsed)); } return JSON.parseObject(JSON.toJSONString(value)); } private JSONArray requiredArray(JSONObject object, String field) { JSONArray array = object.getJSONArray(field); if (array == null) throw new IllegalStateException("Missing BTS simulator state field: " + field); return array; } private int requiredInteger(JSONObject object, String field) { Integer value = object.getInteger(field); if (value == null) throw new IllegalStateException("Missing BTS simulator state field: " + field); return value; } private DeviceState deviceState(Object value) { JSONObject object = jsonObject(value); List channels = new ArrayList<>(); requiredArray(object, "channels").forEach(item -> channels.add(channelState(item))); return new DeviceState(requiredInteger(object, "deviceId"), object.getString("ip"), object.getBooleanValue("online"), channels); } private List embeddedCollectorHistory(JSONObject object) { JSONArray values = object.getJSONArray("history"); if (values == null) return List.of(); List history = new ArrayList<>(values.size()); values.forEach(item -> history.add(collectorMeasurementRecord(item))); return List.copyOf(history); } private CollectorState collectorState(JSONObject object, List embeddedHistory) { 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 channels = new ArrayList<>(); requiredArray(object, "channels").forEach(item -> channels.add(collectorChannelState(item))); List latestMeasurements = new ArrayList<>(); JSONArray latest = object.getJSONArray("latestMeasurements"); if (latest != null) latest.forEach(item -> latestMeasurements.add(collectorRealtimeMeasurement(item))); else 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, historyRecordCount, collectorHistoryFaults(object.get("historyFaults"))); } private CollectorChannelState collectorChannelState(Object value) { JSONObject object = jsonObject(value); List 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 CollectorRealtimeMeasurement collectorRealtimeMeasurement(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 CollectorRealtimeMeasurement(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")); } /** 兼容 Ticket06 开发期间已写入但尚未拆分快照模型的状态。 */ private List latestMeasurementsFromLegacyHistory( List history, List channels) { java.util.Set configured = new java.util.HashSet<>(); channels.forEach(channel -> channel.metrics().forEach(metric -> configured.add(channel.channelCode() + "\u0000" + metric.metricCode()))); java.util.Map latest = new java.util.LinkedHashMap<>(); for (CollectorMeasurementRecord record : history) { String key = record.channelCode() + "\u0000" + record.metricCode(); if (configured.contains(key)) latest.put(key, new CollectorRealtimeMeasurement(record.adapterCode(), record.collectorCode(), record.channelCode(), record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), record.serverReceivedAt(), record.metricCode(), record.value(), record.unit())); } return List.copyOf(latest.values()); } private CollectorHistoryFaults collectorHistoryFaults(Object value) { if (value == null) return CollectorHistoryFaults.none(); JSONObject object = jsonObject(value); List 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) { JSONObject object = jsonObject(value); String workStatus = object.getString("workStatus"); if (workStatus == null) throw new IllegalStateException("Missing BTS simulator state field: workStatus"); return new ChannelState(requiredInteger(object, "channelId"), object.getBooleanValue("online"), DeviceWorkStatus.valueOf(workStatus)); } private ChannelTombstone channelTombstone(Object value) { JSONObject object = jsonObject(value); 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); } private String historyKey(String collectorCode) { return HISTORY_KEY_PREFIX + collectorCode; } private record ParsedSimulatorState(SimulatorState state, List embeddedHistory) { private ParsedSimulatorState { embeddedHistory = List.copyOf(embeddedHistory); } } }