实现 Ticket12 人工结束与顺序接入

yeg

2026-08-30 20b346ef9d4fcbbbe05abfe08e4f4f0303b0812e

 ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java

@@ -4,13 +4,17 @@
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。 */
@@ -19,9 +23,12 @@
{
    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)
    {
@@ -32,18 +39,161 @@
    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);
    }

    @Override
    public SimulatorState resetHistoryForDemoStartup(SimulatorState state)
    {
        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);

        List<com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState> 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)
    {
        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.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<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
@@ -64,11 +214,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 +234,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)
        {
@@ -128,13 +291,33 @@
        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);
                object.getBooleanValue("online"), channels, profile);
    }

    private CollectorState collectorState(Object value)
    private BtsDeviceProfile deviceProfile(JSONObject object)
    {
        JSONObject object = jsonObject(value);
        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");
@@ -151,18 +334,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)
@@ -258,6 +442,13 @@
        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);
@@ -270,4 +461,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); }
    }
}