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

yeg

2026-08-29 d2b53841c31b82df4f71b800ad18a323694c10d7

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

@@ -1,101 +1,133 @@
package com.ruoyi.system.service.impl;

import java.math.BigDecimal;
import java.time.Duration;
import java.time.Instant;
import java.util.LinkedHashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.Set;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;
import com.ruoyi.device.model.ChannelIdentity;
import com.ruoyi.device.model.DetectedChannel;
import com.ruoyi.system.domain.TDetectionCabinet;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TDeviceAdapter;
import com.ruoyi.device.model.DetectedDeviceMetadata;
import com.ruoyi.device.model.DeviceWorkStatus;
import com.ruoyi.device.model.RealtimeSnapshot;
import com.ruoyi.device.registry.DeviceAdapterRegistry;
import com.ruoyi.system.domain.vo.AdapterDiscoveryResult;
import com.ruoyi.system.mapper.TDeviceAdapterMapper;
import com.ruoyi.system.mapper.TDetectionCabinetMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/** 将适配器观测写入 Redis 候选，并同步已登记设备的实时事实。 */
@Service
public class DeviceDiscoveryPersistenceService
{
    private final TDeviceAdapterMapper mapper;
    private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30);
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final TDetectionCabinetMapper detectionCabinetMapper;

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

    /** 在一个事务内 upsert 本轮观测事实；未返回通道保持原记录不变。 */
    @Transactional(rollbackFor = Exception.class)
    public AdapterDiscoveryResult persist(TDeviceAdapter adapter, List<DetectedChannel> detected, Instant observedAt)
    /** 执行设备发现和可用通道查询，并将本轮观测写入 Redis 候选。 */
    public AdapterDiscoveryResult discover(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        Map<CabinetKey, List<DetectedChannel>> cabinets = detected.stream().collect(
        DiscoveryObservation observation = registry.execute(source.adapterCode(), runtime -> {
            List<DetectedChannel> channels = runtime.discoverChannels();
            Set<ChannelIdentity> available = channels.stream()
                    .filter(DetectedChannel::available)
                    .map(DetectedChannel::identity)
                    .collect(java.util.stream.Collectors.toCollection(LinkedHashSet::new));
            Map<ChannelIdentity, RealtimeSnapshot> snapshots = available.isEmpty()
                    ? Map.of() : runtime.inquire(available);
            return new DiscoveryObservation(channels, snapshots);
        });
        Instant observedAt = Instant.now();
        return persist(source, observation.channels(), observation.snapshots(), observedAt);
    }

    /** upsert 本轮发现和实时状态；未返回设备由 expiresAt 控制淘汰。 */
    private AdapterDiscoveryResult persist(DiscoverySource source, List<DetectedChannel> detected,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots, Instant observedAt)
    {
        Map<CabinetKey, List<DetectedChannel>> grouped = detected.stream().collect(
                java.util.stream.Collectors.groupingBy(
                        item -> new CabinetKey(item.identity().devtype(), item.identity().devid()),
                        LinkedHashMap::new, java.util.stream.Collectors.toList()));
        Duration lifetime = Duration.ofMillis(Math.max(MINIMUM_CANDIDATE_LIFETIME.toMillis(),
                (long) source.pollIntervalMs() * 3));
        List<CabinetCandidate> candidates = grouped.entrySet().stream()
                .map(entry -> toCandidate(source, entry.getKey(), entry.getValue(), snapshots,
                        observedAt, observedAt.plus(lifetime)))
                .toList();
        candidateStore.recordCabinets(candidates);
        synchronizeRegisteredFacts(source, detected, snapshots, observedAt);
        return new AdapterDiscoveryResult(grouped.size(), detected.size(), observedAt);
    }

        for (Map.Entry<CabinetKey, List<DetectedChannel>> entry : cabinets.entrySet())
    /**
     * 发现候选与正式设备是两份事实；正式登记后仍需把成功观测同步到 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)
        {
            Long cabinetId = ensureCabinet(adapter, entry.getKey(), entry.getValue(), observedAt);
            for (DetectedChannel item : entry.getValue())
                mapper.upsertDetectedChannel(toChannel(adapter.getAdapterId(), cabinetId, item, observedAt));
            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);
        }
        return new AdapterDiscoveryResult(cabinets.size(), detected.size(), observedAt);
        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 Long ensureCabinet(TDeviceAdapter adapter, CabinetKey key, List<DetectedChannel> channels,
            Instant observedAt)
    private CabinetCandidate toCandidate(DiscoverySource source, CabinetKey key, List<DetectedChannel> channels,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots, Instant observedAt, Instant expiresAt)
    {
        Long cabinetId = mapper.selectCabinetIdByIdentity(adapter.getAdapterId(), key.devtype(), key.devid());
        TDetectionCabinet cabinet = new TDetectionCabinet();
        cabinet.setCabinetId(cabinetId);
        cabinet.setAdapterId(adapter.getAdapterId());
        cabinet.setCabinetCode(adapter.getAdapterCode() + "-" + key.devtype() + "-" + key.devid());
        cabinet.setCabinetName(adapter.getAdapterName() + " " + key.devtype() + "-" + key.devid());
        cabinet.setCapacity(BigDecimal.ONE);
        cabinet.setChannelCount(channels.size());
        cabinet.setVendorDeviceType(key.devtype());
        cabinet.setVendorDeviceId(key.devid());
        cabinet.setIpAddress(channels.stream().map(DetectedChannel::serverIp).filter(value -> value != null && !value.isBlank())
                .findFirst().orElse(null));
        cabinet.setOnlineStatus(channels.stream().anyMatch(DetectedChannel::available) ? "ONLINE" : "OFFLINE");
        cabinet.setLastSeenAtUtc(observedAt);
        if (cabinetId == null)
        {
            try
            {
                mapper.insertDetectedCabinet(cabinet);
                return cabinet.getCabinetId();
            }
            catch (DuplicateKeyException concurrentInsert)
            {
                cabinetId = mapper.selectCabinetIdByIdentity(adapter.getAdapterId(), key.devtype(), key.devid());
                if (cabinetId == null) throw concurrentInsert;
                cabinet.setCabinetId(cabinetId);
            }
        }
        mapper.updateDetectedCabinet(cabinet);
        return cabinetId;
        DetectedDeviceMetadata metadata = channels.get(0).deviceMetadata();
        String name = metadata.name() == null
                ? source.adapterName() + " " + key.devtype() + "-" + key.devid()
                : metadata.name();
        String ip = channels.stream().map(DetectedChannel::serverIp)
                .filter(value -> value != null && !value.isBlank()).findFirst().orElse(null);
        List<ChannelCandidate> channelCandidates = channels.stream().map(item -> {
            RealtimeSnapshot snapshot = snapshots.get(item.identity());
            DeviceWorkStatus status = snapshot == null || snapshot.workStatus() == null
                    ? DeviceWorkStatus.UNKNOWN : snapshot.workStatus();
            return new ChannelCandidate(item.identity(), item.available(), status, observedAt);
        }).toList();
        return new CabinetCandidate(source.adapterCode(), source.adapterId(),
                source.adapterCode() + "-" + key.devtype() + "-" + key.devid(),
                key.devtype(), key.devid(), ip, metadata.macAddress(), name,
                metadata.manufacturer(), metadata.model(), channelCandidates, observedAt, expiresAt);
    }

    private TDetectionChannel toChannel(Long adapterId, Long cabinetId, DetectedChannel detected, Instant observedAt)
    {
        TDetectionChannel channel = new TDetectionChannel();
        channel.setAdapterId(adapterId);
        channel.setCabinetId(cabinetId);
        channel.setDevtype(detected.identity().devtype());
        channel.setDevid(detected.identity().devid());
        channel.setSubdevid(detected.identity().subdevid());
        channel.setChlid(detected.identity().chlid());
        channel.setDisplayNo(detected.identity().subdevid() + "-" + detected.identity().chlid());
        channel.setOnlineStatus(detected.available() ? "ONLINE" : "OFFLINE");
        channel.setWorkStatus("UNKNOWN");
        channel.setLastSeenAtUtc(observedAt);
        channel.setStatus("0");
        return channel;
    }
    /** 发现来源只携带候选标识和过期计算所需配置，不复用数据库实体。 */
    public record DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs) { }

    private record CabinetKey(int devtype, int devid)
    {
    }
    private record DiscoveryObservation(List<DetectedChannel> channels,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots) { }

    private record CabinetKey(int devtype, int devid) { }
}