From 210663040bf65ee00b5f86545fdc31bd0356fddf Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期四, 27 八月 2026 16:31:53 +0800
Subject: [PATCH] feat(device): persist simulator discovery in redis
---
ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java | 266 ++++++++++++++++++++++++++++++++++++++++++----------
1 files changed, 214 insertions(+), 52 deletions(-)
diff --git a/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java b/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java
index 4d78627..c299502 100644
--- a/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java
+++ b/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java
@@ -3,22 +3,29 @@
import java.io.IOException;
import java.io.StringReader;
import java.time.Instant;
-import java.util.ArrayDeque;
import java.util.ArrayList;
-import java.util.Deque;
-import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLStreamConstants;
import javax.xml.stream.XMLStreamReader;
import com.ruoyi.device.model.DeviceWorkStatus;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelTombstone;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.DeviceState;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.RecentRequest;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;
+import com.ruoyi.device.simulator.InMemoryBtsSimulatorStateStore;
/** 提供由开发后端托管的临时只读 BTS 模拟端点。 */
public final class BtsDevelopmentSimulator implements AutoCloseable
{
public static final String HOST = "127.0.0.1";
public static final int DEFAULT_PORT = 15020;
+ public static final String ADAPTER_CODE = "dev-sim";
+ public static final String ADAPTER_NAME = "dev-sim BTS 模拟器";
+ public static final int POLL_INTERVAL_MS = 5000;
private static final int DEVICE_TYPE = 22;
private static final int SUB_DEVICE_ID = 1;
@@ -27,16 +34,22 @@
private static final int TOMBSTONE_DISCOVERY_COUNT = 3;
private final ScriptedBtsTcpSimulator simulator;
- private final Map<Integer, DeviceState> devices = new LinkedHashMap<>();
- private final Map<String, ChannelTombstone> tombstones = new LinkedHashMap<>();
- private final Deque<RecentRequest> recentRequests = new ArrayDeque<>();
- private int nextDeviceId = 1;
+ private final BtsSimulatorStateStore stateStore;
+ private SimulatorState state;
private boolean running = true;
/** 启动仅监听 IPv4 本机回环地址的开发模拟器；传入 0 时由系统分配测试端口。 */
public BtsDevelopmentSimulator(int port) throws IOException
{
- addDeviceInternal("127.0.0.2", 4);
+ this(port, new InMemoryBtsSimulatorStateStore());
+ }
+
+ /** 从长期状态恢复后再启动 TCP 服务，避免服务先暴露默认拓扑。 */
+ public BtsDevelopmentSimulator(int port, BtsSimulatorStateStore stateStore) throws IOException
+ {
+ this.stateStore = stateStore;
+ state = stateStore.loadOrInitialize(defaultState());
+ validateRestoredState();
simulator = ScriptedBtsTcpSimulator.responding(port, this::respond);
}
@@ -49,7 +62,12 @@
public synchronized int addDevice(String ip, int channelCount)
{
validateDeviceConfiguration(ip, channelCount, null);
- return addDeviceInternal(ip, channelCount);
+ int deviceId = state.nextDeviceId();
+ List<DeviceState> devices = new ArrayList<>(state.devices());
+ devices.add(newDevice(deviceId, ip, channelCount));
+ persist(new SimulatorState(BtsSimulatorStateStore.CURRENT_VERSION, devices, state.collectors(),
+ deviceId + 1, state.nextCollectorId(), state.tombstones()));
+ return deviceId;
}
/** 修改开发设备的虚拟地址、通道数和在线状态。 */
@@ -64,20 +82,26 @@
? current.channels().get(channelId - 1)
: new ChannelState(channelId, true, DeviceWorkStatus.STOPPED));
}
+ List<DeviceState> devices = replaceDevice(new DeviceState(deviceId, ip.trim(), online, channels));
+ List<ChannelTombstone> tombstones = new ArrayList<>(state.tombstones());
for (int channelId = channelCount + 1; channelId <= current.channels().size(); channelId++)
- addTombstone(current.deviceId(), current.ip(), channelId);
- for (int channelId = 1; channelId <= channelCount; channelId++)
- tombstones.remove(tombstoneKey(deviceId, channelId));
- devices.put(deviceId, new DeviceState(deviceId, ip.trim(), online, List.copyOf(channels)));
+ upsertTombstone(tombstones, current.deviceId(), current.ip(), channelId);
+ tombstones.removeIf(item -> item.deviceId() == deviceId && item.channelId() <= channelCount);
+ persist(new SimulatorState(state.version(), devices, state.collectors(), state.nextDeviceId(),
+ state.nextCollectorId(), tombstones));
}
/** 删除指定开发设备；后续发现响应不再返回它。 */
public synchronized void deleteDevice(int deviceId)
{
- DeviceState removed = devices.remove(deviceId);
- if (removed == null) throw new IllegalArgumentException("Unknown simulator device: " + deviceId);
+ DeviceState removed = requireDevice(deviceId);
+ List<DeviceState> devices = new ArrayList<>(state.devices());
+ devices.removeIf(item -> item.deviceId() == deviceId);
+ List<ChannelTombstone> tombstones = new ArrayList<>(state.tombstones());
for (ChannelState channel : removed.channels())
- addTombstone(removed.deviceId(), removed.ip(), channel.channelId());
+ upsertTombstone(tombstones, removed.deviceId(), removed.ip(), channel.channelId());
+ persist(new SimulatorState(state.version(), devices, state.collectors(), state.nextDeviceId(),
+ state.nextCollectorId(), tombstones));
}
/** 修改指定开发设备的单个通道状态。 */
@@ -90,30 +114,94 @@
throw new IllegalArgumentException("channelId must be between 1 and " + device.channels().size());
List<ChannelState> channels = new ArrayList<>(device.channels());
channels.set(channelId - 1, new ChannelState(channelId, online, workStatus));
- devices.put(deviceId, new DeviceState(deviceId, device.ip(), device.online(), List.copyOf(channels)));
+ persist(new SimulatorState(state.version(),
+ replaceDevice(new DeviceState(deviceId, device.ip(), device.online(), channels)), state.collectors(),
+ state.nextDeviceId(), state.nextCollectorId(), state.tombstones()));
+ }
+
+ /** 新增一台与 BTS 检测柜拓扑隔离的开发采集器。 */
+ public synchronized int addCollector(String ip, int channelCount)
+ {
+ validateCollectorConfiguration(ip, channelCount, null);
+ int collectorId = state.nextCollectorId();
+ List<CollectorState> collectors = new ArrayList<>(state.collectors());
+ collectors.add(newCollector(collectorId, ip, channelCount));
+ persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
+ collectorId + 1, state.tombstones()));
+ return collectorId;
+ }
+
+ /** 修改开发采集器的虚拟地址、通道数和在线状态。 */
+ public synchronized void updateCollector(int collectorId, String ip, int channelCount, boolean online)
+ {
+ validateCollectorConfiguration(ip, channelCount, collectorId);
+ CollectorState current = requireCollector(collectorId);
+ List<CollectorState> collectors = new ArrayList<>(state.collectors());
+ collectors.replaceAll(item -> item.collectorId() == collectorId
+ ? new CollectorState(collectorId, current.deviceCode(), ip.trim(), current.mac(), current.name(),
+ current.manufacturer(), current.model(), channelCount, online)
+ : item);
+ persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
+ state.nextCollectorId(), state.tombstones()));
+ }
+
+ /** 删除指定开发采集器，不影响检测柜 TCP 发现结果。 */
+ public synchronized void deleteCollector(int collectorId)
+ {
+ requireCollector(collectorId);
+ List<CollectorState> collectors = new ArrayList<>(state.collectors());
+ collectors.removeIf(item -> item.collectorId() == collectorId);
+ persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
+ state.nextCollectorId(), state.tombstones()));
}
/** 返回控制页需要的运行状态快照。 */
public synchronized StateSnapshot snapshot()
{
- List<DeviceSnapshot> deviceSnapshots = devices.values().stream()
+ List<DeviceSnapshot> deviceSnapshots = state.devices().stream()
.map(device -> new DeviceSnapshot(device.deviceId(), device.ip(), device.online(),
device.channels().stream()
.map(channel -> new ChannelSnapshot(channel.channelId(), channel.online(),
channel.workStatus()))
.toList()))
.toList();
- return new StateSnapshot(running, HOST + ":" + port(), deviceSnapshots, List.copyOf(recentRequests));
+ List<CollectorSnapshot> collectorSnapshots = state.collectors().stream()
+ .map(collector -> new CollectorSnapshot(collector.collectorId(), collector.deviceCode(),
+ collector.ip(), collector.mac(), collector.name(), collector.manufacturer(),
+ collector.model(), collector.channelCount(), collector.online()))
+ .toList();
+ return new StateSnapshot(running, HOST + ":" + port(), deviceSnapshots, collectorSnapshots,
+ stateStore.loadRecentRequests());
}
- private int addDeviceInternal(String ip, int channelCount)
+ /** 人工恢复默认拓扑；这是正常运行中唯一清空长期模拟状态的入口。 */
+ public synchronized void reset()
{
- int deviceId = nextDeviceId++;
+ SimulatorState defaultState = defaultState();
+ stateStore.reset(defaultState);
+ state = defaultState;
+ }
+
+ private SimulatorState defaultState()
+ {
+ return new SimulatorState(BtsSimulatorStateStore.CURRENT_VERSION,
+ List.of(newDevice(1, "127.0.0.2", 4)),
+ List.of(newCollector(1, "127.0.1.2", 4)), 2, 2, List.of());
+ }
+
+ private DeviceState newDevice(int deviceId, String ip, int channelCount)
+ {
List<ChannelState> channels = new ArrayList<>(channelCount);
for (int channelId = 1; channelId <= channelCount; channelId++)
channels.add(new ChannelState(channelId, true, DeviceWorkStatus.STOPPED));
- devices.put(deviceId, new DeviceState(deviceId, ip.trim(), true, List.copyOf(channels)));
- return deviceId;
+ return new DeviceState(deviceId, ip.trim(), true, channels);
+ }
+
+ private CollectorState newCollector(int collectorId, String ip, int channelCount)
+ {
+ return new CollectorState(collectorId, "dev-sim-collector-" + collectorId,
+ ip.trim(), simulatorCollectorMac(collectorId), "dev-sim 采集器 " + collectorId,
+ "开发模拟器", "Virtual Collector", channelCount, true);
}
private String respond(String request)
@@ -130,8 +218,11 @@
private synchronized void recordRequest(String command, String request)
{
- recentRequests.addFirst(new RecentRequest(Instant.now(), command, request));
- while (recentRequests.size() > MAX_RECENT_REQUESTS) recentRequests.removeLast();
+ List<RecentRequest> recentRequests = new ArrayList<>(stateStore.loadRecentRequests());
+ recentRequests.add(0, new RecentRequest(Instant.now(), command, request));
+ if (recentRequests.size() > MAX_RECENT_REQUESTS)
+ recentRequests = new ArrayList<>(recentRequests.subList(0, MAX_RECENT_REQUESTS));
+ stateStore.saveRecentRequests(recentRequests);
}
private String readCommand(String request)
@@ -157,11 +248,11 @@
private synchronized String deviceInfoResponse()
{
- int channelCount = devices.values().stream().mapToInt(device -> device.channels().size()).sum()
- + tombstones.size();
+ int channelCount = state.devices().stream().mapToInt(device -> device.channels().size()).sum()
+ + state.tombstones().size();
StringBuilder response = new StringBuilder("<bts version=\"1.0\"><cmd>getdevinfo_resp</cmd>")
.append("<serverip count=\"0\"/><middle count=\"").append(channelCount).append("\">");
- for (DeviceState device : devices.values())
+ for (DeviceState device : state.devices())
{
for (ChannelState channel : device.channels())
{
@@ -169,14 +260,17 @@
appendDiscoveredChannel(response, device.deviceId(), device.ip(), channel.channelId(), available);
}
}
- for (Map.Entry<String, ChannelTombstone> entry : new ArrayList<>(tombstones.entrySet()))
+ List<ChannelTombstone> remaining = new ArrayList<>();
+ for (ChannelTombstone tombstone : state.tombstones())
{
- ChannelTombstone tombstone = entry.getValue();
appendDiscoveredChannel(response, tombstone.deviceId(), tombstone.ip(), tombstone.channelId(), false);
- if (tombstone.remainingDiscoveries() <= 1) tombstones.remove(entry.getKey());
- else tombstones.put(entry.getKey(), new ChannelTombstone(tombstone.deviceId(), tombstone.ip(),
- tombstone.channelId(), tombstone.remainingDiscoveries() - 1));
+ if (tombstone.remainingDiscoveries() > 1)
+ remaining.add(new ChannelTombstone(tombstone.deviceId(), tombstone.ip(), tombstone.channelId(),
+ tombstone.remainingDiscoveries() - 1));
}
+ if (!remaining.equals(state.tombstones()))
+ persist(new SimulatorState(state.version(), state.devices(), state.collectors(), state.nextDeviceId(),
+ state.nextCollectorId(), remaining));
return response.append("</middle></bts>").toString();
}
@@ -195,7 +289,7 @@
private synchronized String observationResponse()
{
List<ObservedChannel> available = new ArrayList<>();
- for (DeviceState device : devices.values())
+ for (DeviceState device : state.devices())
for (ChannelState channel : device.channels())
if (device.online() && channel.online()) available.add(new ObservedChannel(device.deviceId(), channel));
@@ -231,10 +325,23 @@
if (!isIpv4(ip)) throw new IllegalArgumentException("Simulator device IP must be a valid IPv4 address");
if (channelCount < 1 || channelCount > MAX_CHANNEL_COUNT)
throw new IllegalArgumentException("channelCount must be between 1 and " + MAX_CHANNEL_COUNT);
- boolean duplicate = devices.values().stream()
+ boolean duplicate = state.devices().stream()
.anyMatch(device -> device.ip().equals(ip.trim())
&& device.deviceId() != (currentDeviceId == null ? -1 : currentDeviceId));
+ duplicate = duplicate || state.collectors().stream().anyMatch(collector -> collector.ip().equals(ip.trim()));
if (duplicate) throw new IllegalArgumentException("Simulator device IP must be unique");
+ }
+
+ private void validateCollectorConfiguration(String ip, int channelCount, Integer currentCollectorId)
+ {
+ if (!isIpv4(ip)) throw new IllegalArgumentException("Simulator collector IP must be a valid IPv4 address");
+ if (channelCount < 1 || channelCount > MAX_CHANNEL_COUNT)
+ throw new IllegalArgumentException("channelCount must be between 1 and " + MAX_CHANNEL_COUNT);
+ boolean duplicate = state.collectors().stream()
+ .anyMatch(collector -> collector.ip().equals(ip.trim())
+ && collector.collectorId() != (currentCollectorId == null ? -1 : currentCollectorId));
+ duplicate = duplicate || state.devices().stream().anyMatch(device -> device.ip().equals(ip.trim()));
+ if (duplicate) throw new IllegalArgumentException("Simulator collector IP must be unique");
}
private boolean isIpv4(String ip)
@@ -259,26 +366,45 @@
private DeviceState requireDevice(int deviceId)
{
- DeviceState device = devices.get(deviceId);
- if (device == null) throw new IllegalArgumentException("Unknown simulator device: " + deviceId);
- return device;
+ return state.devices().stream().filter(item -> item.deviceId() == deviceId).findFirst()
+ .orElseThrow(() -> new IllegalArgumentException("Unknown simulator device: " + deviceId));
}
- private void addTombstone(int deviceId, String ip, int channelId)
+ private CollectorState requireCollector(int collectorId)
{
- tombstones.put(tombstoneKey(deviceId, channelId),
- new ChannelTombstone(deviceId, ip, channelId, TOMBSTONE_DISCOVERY_COUNT));
+ return state.collectors().stream().filter(item -> item.collectorId() == collectorId).findFirst()
+ .orElseThrow(() -> new IllegalArgumentException("Unknown simulator collector: " + collectorId));
}
- private String tombstoneKey(int deviceId, int channelId)
+ private List<DeviceState> replaceDevice(DeviceState replacement)
{
- return deviceId + ":" + channelId;
+ List<DeviceState> devices = new ArrayList<>(state.devices());
+ devices.replaceAll(item -> item.deviceId() == replacement.deviceId() ? replacement : item);
+ return devices;
+ }
+
+ private void upsertTombstone(List<ChannelTombstone> tombstones, int deviceId, String ip, int channelId)
+ {
+ tombstones.removeIf(item -> item.deviceId() == deviceId && item.channelId() == channelId);
+ tombstones.add(new ChannelTombstone(deviceId, ip, channelId, TOMBSTONE_DISCOVERY_COUNT));
+ }
+
+ private void persist(SimulatorState nextState)
+ {
+ // 写穿失败时保留当前内存副本，避免控制页显示未持久化状态。
+ stateStore.save(nextState);
+ state = nextState;
}
private String simulatorMac(int devtype, int devid)
{
return String.format("02:44:%02X:%02X:%02X:%02X",
(devtype >>> 8) & 0xFF, devtype & 0xFF, (devid >>> 8) & 0xFF, devid & 0xFF);
+ }
+
+ private String simulatorCollectorMac(int collectorId)
+ {
+ return String.format("02:43:00:00:%02X:%02X", (collectorId >>> 8) & 0xFF, collectorId & 0xFF);
}
private void requireControllableStatus(DeviceWorkStatus status)
@@ -288,6 +414,46 @@
throw new IllegalArgumentException("Simulator work status must be STOPPED, WORKING, PAUSED or FINISHED");
}
+ private void validateRestoredState()
+ {
+ java.util.Set<Integer> deviceIds = new java.util.HashSet<>();
+ java.util.Set<Integer> collectorIds = new java.util.HashSet<>();
+ java.util.Set<String> ips = new java.util.HashSet<>();
+ int maximumDeviceId = 0;
+ for (DeviceState device : state.devices())
+ {
+ if (device.deviceId() < 1 || !deviceIds.add(device.deviceId()) || !isIpv4(device.ip())
+ || !ips.add(device.ip()) || device.channels().isEmpty()
+ || device.channels().size() > MAX_CHANNEL_COUNT)
+ throw new IllegalStateException("Invalid BTS simulator device state in Redis");
+ maximumDeviceId = Math.max(maximumDeviceId, device.deviceId());
+ for (int index = 0; index < device.channels().size(); index++)
+ {
+ ChannelState channel = device.channels().get(index);
+ if (channel.channelId() != index + 1)
+ throw new IllegalStateException("Invalid BTS simulator channel state in Redis");
+ requireControllableStatus(channel.workStatus());
+ }
+ }
+ int maximumCollectorId = 0;
+ for (CollectorState collector : state.collectors())
+ {
+ if (collector.collectorId() < 1 || !collectorIds.add(collector.collectorId())
+ || !isIpv4(collector.ip()) || !ips.add(collector.ip())
+ || collector.channelCount() < 1 || collector.channelCount() > MAX_CHANNEL_COUNT)
+ throw new IllegalStateException("Invalid BTS simulator collector state in Redis");
+ maximumCollectorId = Math.max(maximumCollectorId, collector.collectorId());
+ }
+ java.util.Set<String> tombstoneIds = new java.util.HashSet<>();
+ boolean invalidTombstone = state.tombstones().stream().anyMatch(item -> item.deviceId() < 1
+ || !isIpv4(item.ip()) || item.channelId() < 1 || item.channelId() > MAX_CHANNEL_COUNT
+ || item.remainingDiscoveries() < 1 || item.remainingDiscoveries() > TOMBSTONE_DISCOVERY_COUNT
+ || !tombstoneIds.add(item.deviceId() + ":" + item.channelId()));
+ if (state.nextDeviceId() <= maximumDeviceId || state.nextCollectorId() <= maximumCollectorId
+ || invalidTombstone)
+ throw new IllegalStateException("Invalid BTS simulator numbering or tombstone state in Redis");
+ }
+
@Override
public synchronized void close() throws Exception
{
@@ -295,20 +461,16 @@
simulator.close();
}
- private record DeviceState(int deviceId, String ip, boolean online, List<ChannelState> channels) { }
-
- private record ChannelState(int channelId, boolean online, DeviceWorkStatus workStatus) { }
-
private record ObservedChannel(int deviceId, ChannelState channel) { }
- private record ChannelTombstone(int deviceId, String ip, int channelId, int remainingDiscoveries) { }
-
public record StateSnapshot(boolean running, String listenAddress, List<DeviceSnapshot> devices,
- List<RecentRequest> recentRequests) { }
+ List<CollectorSnapshot> collectors, List<RecentRequest> recentRequests) { }
public record DeviceSnapshot(int deviceId, String ip, boolean online, List<ChannelSnapshot> channels) { }
public record ChannelSnapshot(int channelId, boolean online, DeviceWorkStatus workStatus) { }
- public record RecentRequest(Instant receivedAt, String command, String request) { }
+ public record CollectorSnapshot(int collectorId, String deviceCode, String ip, String mac, String name,
+ String manufacturer, String model, int channelCount, boolean online) { }
+
}
--
Gitblit v1.7.1