From 210663040bf65ee00b5f86545fdc31bd0356fddf Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期四, 27 八月 2026 16:31:53 +0800
Subject: [PATCH] feat(device): persist simulator discovery in redis
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceDiscoveryPersistenceService.java | 149 ++++++++++++++++++++++---------------------------
1 files changed, 66 insertions(+), 83 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 b9e9cdf..08582ef 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
@@ -1,114 +1,97 @@
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 com.ruoyi.device.model.DetectedChannel;
+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.device.model.DetectedDeviceMetadata;
+import com.ruoyi.device.model.DeviceWorkStatus;
import com.ruoyi.device.model.RealtimeSnapshot;
-import com.ruoyi.system.domain.TDetectionCabinet;
-import com.ruoyi.system.domain.TDetectionChannel;
-import com.ruoyi.system.domain.TDeviceAdapter;
+import com.ruoyi.device.registry.DeviceAdapterRegistry;
import com.ruoyi.system.domain.vo.AdapterDiscoveryResult;
-import com.ruoyi.system.mapper.TDeviceAdapterMapper;
+import org.springframework.stereotype.Service;
+/** 将适配器观测写入 Redis 候选存储，不创建正式 MySQL 设备或通道。 */
@Service
public class DeviceDiscoveryPersistenceService
{
- private final TDeviceAdapterMapper mapper;
+ private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30);
+ private final DeviceDiscoveryCandidateStore candidateStore;
- public DeviceDiscoveryPersistenceService(TDeviceAdapterMapper mapper)
+ public DeviceDiscoveryPersistenceService(DeviceDiscoveryCandidateStore candidateStore)
{
- this.mapper = mapper;
+ this.candidateStore = candidateStore;
}
- /** 在事务内 upsert 本轮发现和实时观测；未返回通道保持原记录不变。 */
- @Transactional(rollbackFor = Exception.class)
- public AdapterDiscoveryResult persist(TDeviceAdapter adapter, List<DetectedChannel> detected,
+ /** 执行设备发现和可用通道查询，并将本轮观测写入 Redis 候选。 */
+ public AdapterDiscoveryResult discover(DeviceAdapterRegistry registry, DiscoverySource source)
+ {
+ 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>> cabinets = detected.stream().collect(
+ 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()));
-
- for (Map.Entry<CabinetKey, List<DetectedChannel>> entry : cabinets.entrySet())
- {
- Long cabinetId = ensureCabinet(adapter, entry.getKey(), entry.getValue(), observedAt);
- for (DetectedChannel item : entry.getValue())
- mapper.upsertDetectedChannel(toChannel(adapter.getAdapterId(), cabinetId, item,
- snapshots.get(item.identity()), observedAt));
- }
- return new AdapterDiscoveryResult(cabinets.size(), detected.size(), observedAt);
+ 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);
+ return new AdapterDiscoveryResult(grouped.size(), detected.size(), observedAt);
}
- 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());
DetectedDeviceMetadata metadata = channels.get(0).deviceMetadata();
- cabinet.setCabinetName(metadata.name() == null
- ? adapter.getAdapterName() + " " + key.devtype() + "-" + key.devid()
- : metadata.name());
- cabinet.setManufacturer(metadata.manufacturer());
- cabinet.setModel(metadata.model());
- cabinet.setMacAddress(metadata.macAddress());
- 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;
+ 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,
- RealtimeSnapshot snapshot, 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(snapshot == null || snapshot.workStatus() == null
- ? "UNKNOWN" : snapshot.workStatus().name());
- 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) { }
}
--
Gitblit v1.7.1