修复采集器历史状态持续序列化错误

yeg

2026-08-29 370ef4f5e663fdfc1d19d2774dfc4ec1ed0af60e

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
package com.ruoyi.device.simulator;
 
import java.math.BigDecimal;
import java.time.Instant;
import java.util.List;
import com.ruoyi.device.model.DeviceWorkStatus;
 
/** 持久化开发模拟器配置；实现必须在返回前确认写入成功。 */
public interface BtsSimulatorStateStore
{
    int CURRENT_VERSION = 1;
 
    /** 读取已有状态；仅在状态不存在时原子写入并返回默认状态。 */
    SimulatorState loadOrInitialize(SimulatorState defaultState);
 
    /** 完整保存下一版模拟器状态。 */
    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();
 
    /** 保存短期保留的最近请求。 */
    void saveRecentRequests(List<RecentRequest> requests);
 
    /** 人工重置主状态和最近请求。 */
    void reset(SimulatorState defaultState);
 
    record SimulatorState(int version, List<DeviceState> devices, List<CollectorState> collectors,
            int nextDeviceId, int nextCollectorId, List<ChannelTombstone> tombstones)
    {
        public SimulatorState
        {
            if (version != CURRENT_VERSION)
                throw new IllegalStateException("Unsupported BTS simulator state version: " + version);
            devices = List.copyOf(devices);
            collectors = List.copyOf(collectors);
            tombstones = List.copyOf(tombstones);
        }
    }
 
    record DeviceState(int deviceId, String ip, boolean online, List<ChannelState> channels)
    {
        public DeviceState { channels = List.copyOf(channels); }
    }
 
    record CollectorState(int collectorId, String deviceCode, String ip, String mac, String name,
            String manufacturer, String model, int channelCount, boolean online, int sampleIntervalMs,
            long nextSequence, Instant lastSampledAt, List<CollectorChannelState> channels,
            List<CollectorRealtimeMeasurement> latestMeasurements,
            long historyRecordCount, CollectorHistoryFaults historyFaults)
    {
        public CollectorState
        {
            channels = List.copyOf(channels);
            latestMeasurements = List.copyOf(latestMeasurements);
            historyFaults = historyFaults == null ? CollectorHistoryFaults.none() : historyFaults;
        }
 
        /** 兼容只包含发现配置的旧模拟状态；Redis 读取边界会补齐采样合同。 */
        public CollectorState(int collectorId, String deviceCode, String ip, String mac, String name,
                String manufacturer, String model, int channelCount, boolean online)
        {
            this(collectorId, deviceCode, ip, mac, name, manufacturer, model, channelCount, online,
                    DevelopmentCollectorSimulation.DEFAULT_SAMPLE_INTERVAL_MS, 1L, null,
                    DevelopmentCollectorSimulation.defaultChannels(channelCount), List.of(), 0L,
                    CollectorHistoryFaults.none());
        }
    }
 
    record CollectorChannelState(String channelCode, List<CollectorMetricState> metrics)
    {
        public CollectorChannelState { metrics = List.copyOf(metrics); }
    }
 
    record CollectorMetricState(String metricCode, BigDecimal value, String unit) { }
 
    record CollectorRealtimeMeasurement(String adapterCode, String collectorCode, String channelCode,
            long sequenceId, Instant deviceTime, String deviceTimeRaw, Instant serverReceivedAt,
            String metricCode, BigDecimal value, String unit) { }
 
    record CollectorMeasurementRecord(String adapterCode, String collectorCode, String channelCode,
            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)
    {
        public CollectorHistoryFaults
        {
            missingSequences = missingSequences == null ? List.of() : List.copyOf(missingSequences);
        }
 
        public static CollectorHistoryFaults none()
        {
            return new CollectorHistoryFaults(List.of(), null, null, false);
        }
    }
 
    record ChannelState(int channelId, boolean online, DeviceWorkStatus workStatus) { }
 
    record ChannelTombstone(int deviceId, String ip, int channelId, int remainingDiscoveries) { }
 
    record RecentRequest(Instant receivedAt, String command, String request) { }
}