From d2b53841c31b82df4f71b800ad18a323694c10d7 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 29 八月 2026 23:13:40 +0800
Subject: [PATCH] 修复设备发现状态同步与开始接收边界
---
ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java | 7 ++
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java | 40 ++++++++++++
ruoyi-system/src/main/java/com/ruoyi/system/mapper/TCollectorMapper.java | 4 +
ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java | 14 ++++
ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/DeviceAdapterServiceImplTest.java | 31 ++++++++++
ruoyi-system/src/main/resources/mapper/system/TCollectorMapper.xml | 6 ++
ruoyi-system/src/main/resources/mapper/system/TDetectionCabinetMapper.xml | 22 +++++++
ruoyi-system/src/main/java/com/ruoyi/system/mapper/TDetectionCabinetMapper.java | 13 ++++
8 files changed, 133 insertions(+), 4 deletions(-)
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java b/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java
index 395f1fc..22af3a1 100644
--- a/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java
+++ b/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"));
+ }
}
}
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java b/ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java
index cd05a02..16da2ef 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java
+++ b/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();
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/DeviceAdapterServiceImplTest.java b/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/DeviceAdapterServiceImplTest.java
index 8001b0d..4f1d265 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/DeviceAdapterServiceImplTest.java
+++ b/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
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TCollectorMapper.java b/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TCollectorMapper.java
index 2ea8c9b..a92dabc 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TCollectorMapper.java
+++ b/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);
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TDetectionCabinetMapper.java b/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TDetectionCabinetMapper.java
index a53f24e..7105265 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TDetectionCabinetMapper.java
+++ b/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);
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 08582ef..87a4392 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
@@ -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)
{
diff --git a/ruoyi-system/src/main/resources/mapper/system/TCollectorMapper.xml b/ruoyi-system/src/main/resources/mapper/system/TCollectorMapper.xml
index 30b1af1..70e19b1 100644
--- a/ruoyi-system/src/main/resources/mapper/system/TCollectorMapper.xml
+++ b/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>
diff --git a/ruoyi-system/src/main/resources/mapper/system/TDetectionCabinetMapper.xml b/ruoyi-system/src/main/resources/mapper/system/TDetectionCabinetMapper.xml
index d7508de..8ec81b0 100644
--- a/ruoyi-system/src/main/resources/mapper/system/TDetectionCabinetMapper.xml
+++ b/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>
--
Gitblit v1.7.1