From 1522564d7ddf949131641f220766a25d8d981751 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期日, 30 八月 2026 16:07:20 +0800
Subject: [PATCH] 收敛测试暂停继续与停止生命周期
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java | 54 +++++++++++++++++++++++++++++++++++++++++++++++++-----
1 files changed, 49 insertions(+), 5 deletions(-)
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java
index 2dfe562..3ddb1db 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java
+++ b/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)
--
Gitblit v1.7.1