feat(device): persist simulator discovery in redis

yeg

2026-08-27 210663040bf65ee00b5f86545fdc31bd0356fddf

 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) { }

}