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.simulator.BtsSimulatorStateStore.BtsDeviceProfile; 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 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); } @Override public SimulatorState resetHistoryForDemoStartup(SimulatorState state) { pendingEmbeddedHistory = List.of(); Set historyKeys = redisTemplate.opsForSet().members(HISTORY_KEYS_KEY); if (historyKeys != null && !historyKeys.isEmpty()) redisTemplate.delete(historyKeys); redisTemplate.delete(HISTORY_KEYS_KEY); List collectors = state.collectors() .stream() .map(collector -> new com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState( collector.collectorId(), collector.deviceCode(), collector.ip(), collector.mac(), collector.name(), collector.manufacturer(), collector.model(), collector.channelCount(), collector.online(), collector.sampleIntervalMs(), 1L, null, collector.channels(), List.of(), 0L, collector.historyFaults())) .toList(); SimulatorState reset = new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(), state.nextCollectorId(), state.tombstones()); redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(reset)); return reset; } 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.opsForList().trim(historyKey, -MAX_HISTORY_RECORDS, -1); 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 CollectorHistoryBatch loadCollectorHistoryPage(String collectorCode, long offset, int limit, boolean reverse) { if (offset < 0) throw new IllegalArgumentException("offset must not be negative"); if (limit < 1 || limit > MAX_HISTORY_BATCH_SIZE) throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_BATCH_SIZE); String key = historyKey(collectorCode); Long storedSize = redisTemplate.opsForList().size(key); long redisCount = storedSize == null ? 0 : storedSize; List embedded = pendingEmbeddedHistory.stream() .filter(append -> append.collectorCode().equals(collectorCode)) .findFirst().map(CollectorHistoryAppend::records).orElse(List.of()); long total = redisCount + embedded.size(); long from = Math.min(offset, total); int count = (int) Math.min(limit, total - from); List records = new ArrayList<>(count); if (reverse) { long lower = total - from - count; long upper = total - from; long embeddedLower = Math.max(lower, redisCount); for (long index = upper - 1; index >= embeddedLower; index--) records.add(embedded.get((int) (index - redisCount))); long redisUpper = Math.min(upper, redisCount); if (lower < redisUpper) { List stored = redisTemplate.opsForList().range(key, lower, redisUpper - 1); if (stored != null) for (int index = stored.size() - 1; index >= 0; index--) records.add(collectorMeasurementRecord(stored.get(index))); } } else { long upper = from + count; long redisUpper = Math.min(upper, redisCount); if (from < redisUpper) { List stored = redisTemplate.opsForList().range(key, from, redisUpper - 1); if (stored != null) stored.forEach(item -> records.add(collectorMeasurementRecord(item))); } long embeddedFrom = Math.max(from, redisCount); if (embeddedFrom < upper) records.addAll(embedded.subList((int) (embeddedFrom - redisCount), (int) (upper - redisCount))); } if (records.size() != count) throw new IllegalStateException("Collector history page changed while reading Redis"); long nextOffset = from + records.size(); return new CollectorHistoryBatch(offset, nextOffset, nextOffset < total, total, 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))); BtsDeviceProfile profile = object.containsKey("profile") ? deviceProfile(object.getJSONObject("profile")) : null; return new DeviceState(requiredInteger(object, "deviceId"), object.getString("ip"), object.getBooleanValue("online"), channels, profile); } private BtsDeviceProfile deviceProfile(JSONObject object) { if (object == null) throw new IllegalStateException("Invalid BTS simulator device profile"); return new BtsDeviceProfile(requiredInteger(object, "devtype"), requiredInteger(object, "subDeviceId"), object.getString("name"), object.getString("manufacturer"), object.getString("model"), object.getString("serialNumber"), object.getString("firmwareVersion"), requiredInteger(object, "channelCount"), requiredDecimal(object, "currentMinA"), requiredDecimal(object, "currentMaxA"), requiredDecimal(object, "voltageMinV"), requiredDecimal(object, "voltageMaxV"), requiredInteger(object, "sampleIntervalMs")); } 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 BigDecimal requiredDecimal(JSONObject object, String field) { BigDecimal value = object.getBigDecimal(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); } } }