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

yeg

2026-08-29 d2b53841c31b82df4f71b800ad18a323694c10d7

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