修复设备发现状态同步与开始接收边界

yeg

2026-08-29 d2b53841c31b82df4f71b800ad18a323694c10d7

修复设备发现状态同步与开始接收边界

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

@@ -8,6 +8,7 @@
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.CollectorCandidate;
import com.ruoyi.system.mapper.TCollectorMapper;

/** 将模拟采集器的发现和实时状态写入 Redis，不回写正式 MySQL 记录。 */
@Component
@@ -15,12 +16,20 @@
{
    private final BtsDevelopmentSimulator simulator;
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final TCollectorMapper collectorMapper;

    public DevelopmentCollectorStatusSynchronizer(BtsDevelopmentSimulator simulator,
            DeviceDiscoveryCandidateStore candidateStore)
    {
        this(simulator, candidateStore, null);
    }

    public DevelopmentCollectorStatusSynchronizer(BtsDevelopmentSimulator simulator,
            DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper)
    {
        this.simulator = simulator;
        this.candidateStore = candidateStore;
        this.collectorMapper = collectorMapper;
    }

    /** 完整替换 dev-sim 采集器快照，删除配置后候选立即消失。 */
@@ -35,5 +44,10 @@
                        observedAt.plus(Duration.ofSeconds(30))))
                .toList();
        candidateStore.replaceCollectors(BtsDevelopmentSimulator.ADAPTER_CODE, candidates);
        if (collectorMapper != null)
        {
            candidates.forEach(candidate -> collectorMapper.updateRegisteredObservation(candidate.deviceCode(),
                    candidate.online() ? "ONLINE" : "OFFLINE"));
        }
    }
}

 ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java

@@ -4,6 +4,9 @@
import org.junit.jupiter.api.Test;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.discovery.InMemoryDeviceDiscoveryCandidateStore;
import com.ruoyi.system.mapper.TCollectorMapper;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;

class DevelopmentCollectorStatusSynchronizerTest
{
@@ -13,13 +16,15 @@
        try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0))
        {
            InMemoryDeviceDiscoveryCandidateStore store = new InMemoryDeviceDiscoveryCandidateStore();
            TCollectorMapper collectorMapper = mock(TCollectorMapper.class);
            DevelopmentCollectorStatusSynchronizer synchronizer =
                    new DevelopmentCollectorStatusSynchronizer(simulator, store);
                    new DevelopmentCollectorStatusSynchronizer(simulator, store, collectorMapper);
            simulator.updateCollector(1, "127.0.1.2", 4, false);

            synchronizer.synchronizeOnlineStatus();

            assertEquals(false, store.snapshot(java.time.Instant.now()).collectors().get(0).online());
            verify(collectorMapper).updateRegisteredObservation("dev-sim-collector-1", "OFFLINE");

            simulator.deleteCollector(1);
            synchronizer.synchronizeOnlineStatus();

 ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/DeviceAdapterServiceImplTest.java

@@ -39,6 +39,7 @@
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TDeviceAdapter;
import com.ruoyi.system.mapper.TDeviceAdapterMapper;
import com.ruoyi.system.mapper.TDetectionCabinetMapper;
import com.ruoyi.system.service.DeviceAdapterErrorCodes;

@ExtendWith(MockitoExtension.class)
@@ -53,6 +54,9 @@
    @Mock
    private DeviceAdapter deviceAdapter;

    @Mock
    private TDetectionCabinetMapper detectionCabinetMapper;

    private DeviceAdapterServiceImpl service;
    private TDeviceAdapter adapter;
    private InMemoryDeviceDiscoveryCandidateStore candidateStore;
@@ -62,7 +66,7 @@
    {
        candidateStore = new InMemoryDeviceDiscoveryCandidateStore();
        service = new DeviceAdapterServiceImpl(adapterMapper, registry,
                new DeviceDiscoveryPersistenceService(candidateStore), candidateStore);
                new DeviceDiscoveryPersistenceService(candidateStore, detectionCabinetMapper), candidateStore);
        adapter = new TDeviceAdapter();
        adapter.setAdapterId(7L);
        adapter.setAdapterCode("lab-bts");
@@ -162,6 +166,10 @@

        var channel = candidateStore.snapshot(Instant.now()).cabinets().get(0).channels().get(0);
        assertEquals(DeviceWorkStatus.WORKING, channel.workStatus());
        verify(detectionCabinetMapper).updateRegisteredChannelObservation(eq("lab-bts"), eq(22), eq(1), eq(2),
                eq(3), eq("ONLINE"), eq("WORKING"), any(), eq(true));
        verify(detectionCabinetMapper).updateRegisteredCabinetObservation(eq("lab-bts"), eq(22), eq(1),
                eq("ONLINE"), any(), eq(true));
    }

    @Test
@@ -179,6 +187,27 @@
        verify(deviceAdapter, never()).inquire(any());
        assertEquals(false, cabinet.online());
        assertTrue(cabinet.channels().stream().noneMatch(channel -> channel.online()));
        verify(detectionCabinetMapper).updateRegisteredChannelObservation(eq("lab-bts"), eq(22), eq(1), eq(1),
                eq(1), eq("OFFLINE"), eq("UNKNOWN"), any(), eq(false));
        verify(detectionCabinetMapper).updateRegisteredCabinetObservation(eq("lab-bts"), eq(22), eq(1),
                eq("OFFLINE"), any(), eq(false));
    }

    @Test
    void availableChannelWithoutSuccessfulSnapshotCannotAdvanceLastSeenBoundary()
    {
        ChannelIdentity identity = new ChannelIdentity("lab-bts", 22, 1, 1, 1);
        when(adapterMapper.selectDeviceAdapterById(7L)).thenReturn(adapter);
        when(deviceAdapter.discoverChannels()).thenReturn(List.of(new DetectedChannel(identity, "127.0.0.1", true)));
        when(deviceAdapter.inquire(Set.of(identity))).thenReturn(Map.of());
        stubRegistryExecution();

        service.discover(7L);

        verify(detectionCabinetMapper).updateRegisteredChannelObservation(eq("lab-bts"), eq(22), eq(1), eq(1),
                eq(1), eq("ONLINE"), eq("UNKNOWN"), any(), eq(false));
        verify(detectionCabinetMapper).updateRegisteredCabinetObservation(eq("lab-bts"), eq(22), eq(1),
                eq("ONLINE"), any(), eq(false));
    }

    @Test

 ruoyi-system/src/main/java/com/ruoyi/system/mapper/TCollectorMapper.java

@@ -29,6 +29,10 @@
    /** 新增采集器。 */
    int insertCollector(TCollector collector);

    /** 同步正式采集器的实时在线事实，不触碰软删除记录。 */
    int updateRegisteredObservation(@Param("collectorCode") String collectorCode,
            @Param("onlineStatus") String onlineStatus);

    /** 编辑采集器基础资料。 */
    int updateCollector(TCollector collector);


 ruoyi-system/src/main/java/com/ruoyi/system/mapper/TDetectionCabinetMapper.java

@@ -61,6 +61,19 @@
    /** 写入正式登记检测柜的稳定通道身份。 */
    int insertDetectionChannel(com.ruoyi.system.domain.TDetectionChannel channel);

    /** 同步正式检测柜的最近发现事实，不触碰软删除记录。 */
    int updateRegisteredCabinetObservation(@Param("adapterCode") String adapterCode,
            @Param("devtype") int devtype, @Param("devid") int devid,
            @Param("onlineStatus") String onlineStatus, @Param("lastSeenAtUtc") java.time.Instant lastSeenAtUtc,
            @Param("updateLastSeen") boolean updateLastSeen);

    /** 按检测柜通道复合身份同步最近发现事实，不触碰软删除记录。 */
    int updateRegisteredChannelObservation(@Param("adapterCode") String adapterCode,
            @Param("devtype") int devtype, @Param("devid") int devid,
            @Param("subdevid") int subdevid, @Param("chlid") int chlid,
            @Param("onlineStatus") String onlineStatus, @Param("workStatus") String workStatus,
            @Param("lastSeenAtUtc") java.time.Instant lastSeenAtUtc, @Param("updateLastSeen") boolean updateLastSeen);

    /** 软删除正式检测柜对应的通道拓扑。 */
    int deleteDetectionChannelsByCabinetId(@Param("cabinetId") Long cabinetId);


 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java

@@ -17,18 +17,24 @@
import com.ruoyi.device.model.RealtimeSnapshot;
import com.ruoyi.device.registry.DeviceAdapterRegistry;
import com.ruoyi.system.domain.vo.AdapterDiscoveryResult;
import com.ruoyi.system.mapper.TDetectionCabinetMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/** 将适配器观测写入 Redis 候选存储，不创建正式 MySQL 设备或通道。 */
/** 将适配器观测写入 Redis 候选，并同步已登记设备的实时事实。 */
@Service
public class DeviceDiscoveryPersistenceService
{
    private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30);
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final TDetectionCabinetMapper detectionCabinetMapper;

    public DeviceDiscoveryPersistenceService(DeviceDiscoveryCandidateStore candidateStore)
    @Autowired
    public DeviceDiscoveryPersistenceService(DeviceDiscoveryCandidateStore candidateStore,
            TDetectionCabinetMapper detectionCabinetMapper)
    {
        this.candidateStore = candidateStore;
        this.detectionCabinetMapper = detectionCabinetMapper;
    }

    /** 执行设备发现和可用通道查询，并将本轮观测写入 Redis 候选。 */
@@ -63,9 +69,39 @@
                        observedAt, observedAt.plus(lifetime)))
                .toList();
        candidateStore.recordCabinets(candidates);
        synchronizeRegisteredFacts(source, detected, snapshots, observedAt);
        return new AdapterDiscoveryResult(grouped.size(), detected.size(), observedAt);
    }

    /**
     * 发现候选与正式设备是两份事实；正式登记后仍需把成功观测同步到 MySQL，
     * 否则开始接收无法取得可靠边界。离线轮询只更新当前状态，不覆盖最后成功观测时间。
     */
    private void synchronizeRegisteredFacts(DiscoverySource source, List<DetectedChannel> detected,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots, Instant observedAt)
    {
        Map<CabinetKey, Boolean> cabinetOnline = new LinkedHashMap<>();
        for (DetectedChannel channel : detected)
        {
            ChannelIdentity identity = channel.identity();
            boolean successfulObservation = channel.available() && snapshots.containsKey(identity);
            RealtimeSnapshot snapshot = snapshots.get(identity);
            DeviceWorkStatus workStatus = snapshot == null || snapshot.workStatus() == null
                    ? DeviceWorkStatus.UNKNOWN : snapshot.workStatus();
            detectionCabinetMapper.updateRegisteredChannelObservation(source.adapterCode(), identity.devtype(),
                    identity.devid(), identity.subdevid(), identity.chlid(), channel.available() ? "ONLINE" : "OFFLINE",
                    workStatus.name(), observedAt, successfulObservation);
            CabinetKey cabinet = new CabinetKey(identity.devtype(), identity.devid());
            cabinetOnline.merge(cabinet, channel.available(), Boolean::logicalOr);
        }
        cabinetOnline.forEach((cabinet, online) -> detectionCabinetMapper.updateRegisteredCabinetObservation(
                source.adapterCode(), cabinet.devtype(), cabinet.devid(), online ? "ONLINE" : "OFFLINE", observedAt,
                online && detected.stream().anyMatch(channel -> channel.available()
                        && channel.identity().devtype() == cabinet.devtype()
                        && channel.identity().devid() == cabinet.devid()
                        && snapshots.containsKey(channel.identity()))));
    }

    private CabinetCandidate toCandidate(DiscoverySource source, CabinetKey key, List<DetectedChannel> channels,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots, Instant observedAt, Instant expiresAt)
    {

 ruoyi-system/src/main/resources/mapper/system/TCollectorMapper.xml

@@ -100,6 +100,12 @@
         #{status}, #{delFlag}, #{createBy}, #{createTime}, #{remark})
    </insert>

    <update id="updateRegisteredObservation">
        update t_collector
        set online_status = #{onlineStatus}, update_time = sysdate()
        where collector_code = #{collectorCode} and del_flag = 0
    </update>

    <update id="updateCollector" parameterType="TCollector">
        update t_collector
        <set>

 ruoyi-system/src/main/resources/mapper/system/TDetectionCabinetMapper.xml

@@ -109,6 +109,28 @@
         #{onlineStatus}, #{workStatus}, #{lastSeenAtUtc}, '0', sysdate(), 0)
    </insert>

    <update id="updateRegisteredCabinetObservation">
        update t_detection_cabinet
        set online_status = #{onlineStatus}, update_time = sysdate()
        <if test="updateLastSeen">, last_seen_at_utc = #{lastSeenAtUtc}</if>
        where adapter_code = #{adapterCode}
          and vendor_device_type = #{devtype} and vendor_device_id = #{devid}
          and del_flag = 0 and registration_status = 'REGISTERED'
    </update>

    <update id="updateRegisteredChannelObservation">
        update t_detection_channel
        set online_status = #{onlineStatus}, work_status = #{workStatus}, update_time = sysdate()
        <if test="updateLastSeen">, last_seen_at_utc = #{lastSeenAtUtc}</if>
        where adapter_code = #{adapterCode}
          and devtype = #{devtype} and devid = #{devid}
          and subdevid = #{subdevid} and chlid = #{chlid}
          and del_flag = 0
          and exists (select 1 from t_detection_cabinet cabinet
                      where cabinet.cabinet_id = t_detection_channel.cabinet_id
                        and cabinet.del_flag = 0 and cabinet.registration_status = 'REGISTERED')
    </update>

    <update id="updateDetectionCabinet" parameterType="TDetectionCabinet">
        update t_detection_cabinet
        <set>