收敛测试暂停继续与停止生命周期

yeg

2026-08-30 1522564d7ddf949131641f220766a25d8d981751

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

@@ -16,8 +16,10 @@
import com.ruoyi.device.model.DeviceWorkStatus;
import com.ruoyi.device.model.RealtimeSnapshot;
import com.ruoyi.device.registry.DeviceAdapterRegistry;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.vo.AdapterDiscoveryResult;
import com.ruoyi.system.mapper.TDetectionCabinetMapper;
import com.ruoyi.system.mapper.TTestMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@@ -28,17 +30,45 @@
    private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30);
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final TDetectionCabinetMapper detectionCabinetMapper;
    private final TTestMapper testMapper;

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

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

    /** 执行设备发现和可用通道查询，并将本轮观测写入 Redis 候选。 */
    public AdapterDiscoveryResult discover(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        return discoverNow(registry, source);
    }

    /** 调度器观测；暂停或停止测试且没有其他接收测试时，不再向该适配器请求实时数据。 */
    public AdapterDiscoveryResult discoverForReception(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        if (!shouldPollReception(source.adapterCode()))
            return new AdapterDiscoveryResult(0, 0, Instant.now());
        return discoverNow(registry, source);
    }

    /** 判断适配器是否仍需为活动测试发起实时请求。 */
    public boolean shouldPollReception(String adapterCode)
    {
        return testMapper == null || testMapper.hasReceivingTestForAdapter(adapterCode) > 0
                || testMapper.hasPausedOrEndedTestForAdapter(adapterCode) == 0;
    }

    private AdapterDiscoveryResult discoverNow(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        DiscoveryObservation observation = registry.execute(source.adapterCode(), runtime -> {
            List<DetectedChannel> channels = runtime.discoverChannels();
@@ -46,8 +76,9 @@
                    .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);
            Set<ChannelIdentity> requested = receivingIdentities(source.adapterCode(), available);
            Map<ChannelIdentity, RealtimeSnapshot> snapshots = requested.isEmpty()
                    ? Map.of() : runtime.inquire(requested);
            return new DiscoveryObservation(channels, snapshots);
        });
        Instant observedAt = Instant.now();
@@ -73,9 +104,22 @@
        return new AdapterDiscoveryResult(grouped.size(), detected.size(), observedAt);
    }

    private Set<ChannelIdentity> receivingIdentities(String adapterCode, Set<ChannelIdentity> available)
    {
        if (testMapper == null) return available;
        List<TDetectionChannel> receiving = testMapper.selectReceivingChannelsForAdapter(adapterCode);
        if (receiving == null || receiving.isEmpty()) return Set.of();
        Set<ChannelIdentity> identities = receiving.stream()
                .map(channel -> new ChannelIdentity(adapterCode, channel.getDevtype(), channel.getDevid(),
                        channel.getSubdevid(), channel.getChlid()))
                .collect(java.util.stream.Collectors.toCollection(LinkedHashSet::new));
        identities.retainAll(available);
        return identities;
    }

    /**
     * 发现候选与正式设备是两份事实；正式登记后仍需把成功观测同步到 MySQL，
     * 否则开始接收无法取得可靠边界。离线轮询只更新当前状态，不覆盖最后成功观测时间。
     * 发现候选与正式设备是两份事实；正式登记后仍需把成功观测同步到 MySQL。
     * 离线轮询只更新当前状态，不覆盖最后成功观测时间；暂停期间不会进入此路径。
     */
    private void synchronizeRegisteredFacts(DiscoverySource source, List<DetectedChannel> detected,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots, Instant observedAt)