package com.battery.business.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.battery.device.api.discovery.DeviceDiscoveryCandidateStore;
|
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
|
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;
|
import com.battery.device.api.model.ChannelIdentity;
|
import com.battery.device.api.model.DetectedChannel;
|
import com.battery.device.api.model.DetectedDeviceMetadata;
|
import com.battery.device.api.model.DeviceWorkStatus;
|
import com.battery.device.api.model.RealtimeSnapshot;
|
import com.battery.device.api.registry.DeviceAdapterRegistry;
|
import com.battery.business.domain.TDetectionChannel;
|
import com.battery.business.domain.vo.AdapterDiscoveryResult;
|
import com.battery.business.mapper.TDetectionCabinetMapper;
|
import com.battery.business.mapper.TTestMapper;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.stereotype.Service;
|
|
/** 将适配器观测写入 Redis 候选，并同步已登记设备的实时事实。 */
|
@Service
|
public class DeviceDiscoveryPersistenceService
|
{
|
private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30);
|
/**
|
* 直连厂商存储的来源，其通道登记清单是长期存在的事实，没有"过期"的对应语义：
|
* 候选是否出现在界面上只由"已登记且未删除"决定，这个时间只是让候选一直有效。
|
*/
|
static final Instant CANDIDATE_NEVER_EXPIRES = Instant.parse("9999-12-31T23:59:59Z");
|
private final DeviceDiscoveryCandidateStore candidateStore;
|
private final TDetectionCabinetMapper detectionCabinetMapper;
|
private final TTestMapper testMapper;
|
|
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)
|
{
|
return discoverNow(registry, source);
|
}
|
|
/** 判断适配器是否存在需要实时请求的接收中测试。 */
|
public boolean shouldPollReception(String adapterCode)
|
{
|
return testMapper == null || testMapper.hasReceivingTestForAdapter(adapterCode) > 0;
|
}
|
|
/** 发现失败或返回空拓扑时，清除该适配器下已登记设备的过期运行事实。 */
|
public void markUnavailable(String adapterCode)
|
{
|
Instant failedAt = Instant.now();
|
// 失败立即撤下旧测量，但保留来源身份，不能等候选 TTL 到期才停止展示成功值。
|
candidateStore.recordCabinets(candidateStore.snapshot(failedAt).cabinets().stream()
|
.filter(item -> item.adapterCode().equals(adapterCode))
|
.map(item -> new CabinetCandidate(item.adapterCode(), item.adapterId(), item.deviceCode(),
|
item.devtype(), item.devid(), item.ip(), item.mac(), item.name(), item.manufacturer(), item.model(),
|
item.channels().stream().map(channel -> new ChannelCandidate(channel.identity(), false,
|
DeviceWorkStatus.UNOBSERVABLE, failedAt, null, channel.lastDataAt(),
|
channel.sourceDisplayNo())).toList(),
|
failedAt, item.expiresAt(), item.lastDataAt()))
|
.toList());
|
detectionCabinetMapper.markRegisteredFactsUnobservable(adapterCode);
|
detectionCabinetMapper.markRegisteredCabinetsOffline(adapterCode);
|
}
|
|
private AdapterDiscoveryResult discoverNow(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));
|
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();
|
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));
|
Instant expiresAt = source.persistentCandidates() ? CANDIDATE_NEVER_EXPIRES : observedAt.plus(lifetime);
|
List<CabinetCandidate> candidates = grouped.entrySet().stream()
|
.map(entry -> toCandidate(source, entry.getKey(), entry.getValue(), snapshots,
|
observedAt, expiresAt))
|
.toList();
|
candidateStore.recordCabinets(candidates);
|
if (detected.isEmpty()) markUnavailable(source.adapterCode());
|
synchronizeRegisteredFacts(source, detected, snapshots, observedAt);
|
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。
|
* 暂停期间仍同步设备状态，但不请求该测试的测量；离线不覆盖最后成功观测时间。
|
*/
|
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.sourceDisplayNo(),
|
channel.available() ? "ONLINE" : "OFFLINE",
|
workStatus.name(), observedAt, successfulObservation);
|
if (channel.available() && !successfulObservation)
|
detectionCabinetMapper.touchRegisteredChannelObservation(source.adapterCode(), identity.devtype(),
|
identity.devid(), identity.subdevid(), identity.chlid(), observedAt);
|
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()))));
|
cabinetOnline.forEach((cabinet, online) -> {
|
boolean successfulSnapshot = online && detected.stream().anyMatch(channel -> channel.available()
|
&& channel.identity().devtype() == cabinet.devtype()
|
&& channel.identity().devid() == cabinet.devid()
|
&& snapshots.containsKey(channel.identity()));
|
if (online && !successfulSnapshot)
|
detectionCabinetMapper.touchRegisteredCabinetObservation(source.adapterCode(), cabinet.devtype(),
|
cabinet.devid(), 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,
|
snapshots.get(item.identity()), item.lastDataAt(), item.sourceDisplayNo());
|
}).toList();
|
Instant lastDataAt = channels.stream().map(DetectedChannel::lastDataAt)
|
.filter(java.util.Objects::nonNull).max(Instant::compareTo).orElse(null);
|
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, lastDataAt);
|
}
|
|
/** 发现来源只携带候选标识和过期计算所需配置，不复用数据库实体。 */
|
public record DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs,
|
boolean persistentCandidates)
|
{
|
/** 兼容按轮询过期计算候选的既有来源。 */
|
public DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs)
|
{
|
this(adapterCode, adapterId, adapterName, pollIntervalMs, false);
|
}
|
}
|
|
private record DiscoveryObservation(List<DetectedChannel> channels,
|
Map<ChannelIdentity, RealtimeSnapshot> snapshots) { }
|
|
private record CabinetKey(int devtype, int devid) { }
|
}
|
