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<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>> 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<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 CabinetCandidate toCandidate(DiscoverySource source, CabinetKey key, List<DetectedChannel> channels,
|
Map<ChannelIdentity, RealtimeSnapshot> 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<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);
|
}
|
|
/** 发现来源只携带候选标识和过期计算所需配置，不复用数据库实体。 */
|
public record DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs) { }
|
|
private record DiscoveryObservation(List<DetectedChannel> channels,
|
Map<ChannelIdentity, RealtimeSnapshot> snapshots) { }
|
|
private record CabinetKey(int devtype, int devid) { }
|
}
|
