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<Object, Object> redisTemplate;
|
private List<CollectorHistoryAppend> pendingEmbeddedHistory = List.of();
|
|
public RedisBtsSimulatorStateStore(RedisTemplate<Object, Object> 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<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 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<CollectorMeasurementRecord> 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<CollectorMeasurementRecord> 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<Object> 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<Object> 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<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
|
public List<RecentRequest> 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<RecentRequest> 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<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 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<DeviceState> devices = new ArrayList<>();
|
requiredArray(root, "devices").forEach(item -> devices.add(deviceState(item)));
|
List<CollectorState> collectors = new ArrayList<>();
|
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)));
|
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<ChannelState> 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<CollectorMeasurementRecord> embeddedCollectorHistory(JSONObject object)
|
{
|
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");
|
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<CollectorRealtimeMeasurement> 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<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 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<CollectorRealtimeMeasurement> latestMeasurementsFromLegacyHistory(
|
List<CollectorMeasurementRecord> history, List<CollectorChannelState> channels)
|
{
|
java.util.Set<String> configured = new java.util.HashSet<>();
|
channels.forEach(channel -> channel.metrics().forEach(metric ->
|
configured.add(channel.channelCode() + "\u0000" + metric.metricCode())));
|
java.util.Map<String, CollectorRealtimeMeasurement> 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<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)
|
{
|
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<CollectorHistoryAppend> embeddedHistory)
|
{
|
private ParsedSimulatorState { embeddedHistory = List.copyOf(embeddedHistory); }
|
}
|
}
|
