package com.battery.business.runtime; import java.time.Instant; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore; import com.battery.device.api.model.ChannelIdentity; import com.battery.device.api.model.DeviceWorkStatus; import com.battery.device.api.model.RealtimeSnapshot; /** 保存发现候选和实时状态;条目按 expiresAt 逻辑过期,Redis key 本身不设短 TTL。 */ @Component public class RedisDeviceDiscoveryCandidateStore implements DeviceDiscoveryCandidateStore { static final String STATE_KEY = "dev:device-discovery:candidates:v1"; private static final int CURRENT_VERSION = 1; private final RedisTemplate redisTemplate; public RedisDeviceDiscoveryCandidateStore(RedisTemplate redisTemplate) { this.redisTemplate = redisTemplate; } /** 按设备编码 upsert 本轮检测柜候选,本轮未返回的候选保留至逻辑过期。 */ @Override public synchronized void recordCabinets(List candidates) { StoredState current = load(); Map merged = new LinkedHashMap<>(); current.cabinets().forEach(item -> merged.put(item.deviceCode(), item)); candidates.forEach(item -> merged.put(item.deviceCode(), item)); save(new StoredState(CURRENT_VERSION, List.copyOf(merged.values()), current.collectors())); } /** 替换指定设备适配器实例的全部采集器候选,其它实例的候选保持不变。 */ @Override public synchronized void replaceCollectors(String adapterCode, List candidates) { StoredState current = load(); List retained = current.collectors().stream() .filter(item -> !item.adapterCode().equals(adapterCode)).toList(); java.util.ArrayList merged = new java.util.ArrayList<>(retained); merged.addAll(candidates); save(new StoredState(CURRENT_VERSION, current.cabinets(), merged)); } /** 删除指定设备适配器实例的检测柜与采集器候选;仅供人工重置或实机联调清理。 */ @Override public synchronized void removeAdapter(String adapterCode) { StoredState current = load(); List cabinets = current.cabinets().stream() .filter(item -> !item.adapterCode().equals(adapterCode)).toList(); List collectors = current.collectors().stream() .filter(item -> !item.adapterCode().equals(adapterCode)).toList(); save(new StoredState(CURRENT_VERSION, cabinets, collectors)); } /** * 返回指定时刻尚未过期的发现候选,并顺带清掉已逻辑过期的条目。 * *

是否“在用”由操作员判断,适配器不自动判定。

*/ @Override public synchronized CandidateSnapshot snapshot(Instant now) { StoredState current = load(); List cabinets = current.cabinets().stream() .filter(item -> item.expiresAt().isAfter(now)).toList(); List collectors = current.collectors().stream() .filter(item -> item.expiresAt().isAfter(now)).toList(); if (cabinets.size() != current.cabinets().size() || collectors.size() != current.collectors().size()) save(new StoredState(CURRENT_VERSION, cabinets, collectors)); return new CandidateSnapshot(cabinets, collectors); } private StoredState load() { Object stored = redisTemplate.opsForValue().get(STATE_KEY); if (stored == null) return new StoredState(CURRENT_VERSION, List.of(), List.of()); try { // 通用 RedisTemplate 可能把泛型数组恢复为 JSONObject;在存储边界逐字段重建,禁止业务层强转。 JSONObject root = jsonObject(stored); Integer version = root.getInteger("version"); if (version == null || version != CURRENT_VERSION) throw new IllegalStateException("Unsupported device discovery state version: " + version); List cabinets = new ArrayList<>(); requiredArray(root, "cabinets").forEach(item -> cabinets.add(cabinetCandidate(item))); List collectors = new ArrayList<>(); requiredArray(root, "collectors").forEach(item -> collectors.add(collectorCandidate(item))); return new StoredState(CURRENT_VERSION, cabinets, collectors); } catch (IllegalStateException exception) { throw exception; } catch (Exception exception) { throw new IllegalStateException("Invalid device discovery state in Redis", exception); } } private JSONObject jsonObject(Object value) { if (value instanceof JSONObject object) return object; if (value instanceof String text) { Object parsed = JSON.parse(text); if (parsed instanceof String nested) parsed = JSON.parse(nested); if (parsed instanceof JSONObject object) return object; return JSON.parseObject(JSON.toJSONString(parsed)); } return JSON.parseObject(JSON.toJSONString(value)); } private JSONArray requiredArray(JSONObject object, String field) { JSONArray array = object.getJSONArray(field); if (array == null) throw new IllegalStateException("Missing device discovery state field: " + field); return array; } private CabinetCandidate cabinetCandidate(Object value) { JSONObject object = jsonObject(value); List channels = new ArrayList<>(); requiredArray(object, "channels").forEach(item -> channels.add(channelCandidate(item))); return new CabinetCandidate(object.getString("adapterCode"), object.getLong("adapterId"), object.getString("deviceCode"), object.getIntValue("devtype"), object.getIntValue("devid"), object.getString("ip"), object.getString("mac"), object.getString("name"), object.getString("manufacturer"), object.getString("model"), channels, requiredInstant(object, "observedAt"), requiredInstant(object, "expiresAt"), optionalInstant(object, "lastDataAt")); } private ChannelCandidate channelCandidate(Object value) { JSONObject object = jsonObject(value); JSONObject identity = jsonObject(object.get("identity")); ChannelIdentity channelIdentity = new ChannelIdentity(identity.getString("adapterCode"), identity.getIntValue("devtype"), identity.getIntValue("devid"), identity.getIntValue("subdevid"), identity.getIntValue("chlid")); RealtimeSnapshot latestSnapshot = object.getObject("latestSnapshot", RealtimeSnapshot.class); // sourceDisplayNo 用宽容读法:升级前写进 Redis 的候选没有这个键,读成空即可,不该报错。 return new ChannelCandidate(channelIdentity, object.getBooleanValue("online"), DeviceWorkStatus.valueOf(object.getString("workStatus")), requiredInstant(object, "observedAt"), latestSnapshot, optionalInstant(object, "lastDataAt"), object.getString("sourceDisplayNo")); } private CollectorCandidate collectorCandidate(Object value) { JSONObject object = jsonObject(value); return new CollectorCandidate(object.getString("adapterCode"), object.getString("deviceCode"), object.getString("ip"), object.getString("mac"), object.getString("name"), object.getString("manufacturer"), object.getString("model"), object.getIntValue("channelCount"), object.getBooleanValue("online"), requiredInstant(object, "observedAt"), requiredInstant(object, "expiresAt")); } private Instant requiredInstant(JSONObject object, String field) { Instant value = object.getObject(field, Instant.class); if (value == null) throw new IllegalStateException("Missing device discovery state field: " + field); return value; } /** 最近数据时间只在厂商存储来源上存在,缺失是正常取值。 */ private Instant optionalInstant(JSONObject object, String field) { return object.getObject(field, Instant.class); } private void save(StoredState state) { redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(state)); } private record StoredState(int version, List cabinets, List collectors) { private StoredState { cabinets = List.copyOf(cabinets); collectors = List.copyOf(collectors); } } }