package com.ruoyi.system.service.impl; 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 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.device.registry.DeviceAdapterRegistry; import com.ruoyi.system.domain.vo.AdapterDiscoveryResult; import org.springframework.stereotype.Service; /** 将适配器观测写入 Redis 候选存储,不创建正式 MySQL 设备或通道。 */ @Service public class DeviceDiscoveryPersistenceService { private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30); private final DeviceDiscoveryCandidateStore candidateStore; public DeviceDiscoveryPersistenceService(DeviceDiscoveryCandidateStore candidateStore) { this.candidateStore = candidateStore; } /** 执行设备发现和可用通道查询,并将本轮观测写入 Redis 候选。 */ public AdapterDiscoveryResult discover(DeviceAdapterRegistry registry, DiscoverySource source) { DiscoveryObservation observation = registry.execute(source.adapterCode(), runtime -> { List channels = runtime.discoverChannels(); Set available = channels.stream() .filter(DetectedChannel::available) .map(DetectedChannel::identity) .collect(java.util.stream.Collectors.toCollection(LinkedHashSet::new)); Map 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 detected, Map snapshots, Instant observedAt) { Map> 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 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 CabinetCandidate toCandidate(DiscoverySource source, CabinetKey key, List channels, Map snapshots, Instant observedAt, Instant expiresAt) { 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 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); } /** 发现来源只携带候选标识和过期计算所需配置,不复用数据库实体。 */ public record DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs) { } private record DiscoveryObservation(List channels, Map snapshots) { } private record CabinetKey(int devtype, int devid) { } }