| | |
| | | Map<String, CabinetCandidate> 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())); |
| | | save(new StoredState(CURRENT_VERSION, List.copyOf(merged.values()))); |
| | | } |
| | | |
| | | /** 替换指定设备适配器实例的全部采集器候选,其它实例的候选保持不变。 */ |
| | | @Override |
| | | public synchronized void replaceCollectors(String adapterCode, List<CollectorCandidate> candidates) |
| | | { |
| | | StoredState current = load(); |
| | | List<CollectorCandidate> retained = current.collectors().stream() |
| | | .filter(item -> !item.adapterCode().equals(adapterCode)).toList(); |
| | | java.util.ArrayList<CollectorCandidate> 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<CabinetCandidate> cabinets = current.cabinets().stream() |
| | | .filter(item -> !item.adapterCode().equals(adapterCode)).toList(); |
| | | List<CollectorCandidate> collectors = current.collectors().stream() |
| | | .filter(item -> !item.adapterCode().equals(adapterCode)).toList(); |
| | | save(new StoredState(CURRENT_VERSION, cabinets, collectors)); |
| | | save(new StoredState(CURRENT_VERSION, cabinets)); |
| | | } |
| | | |
| | | /** |
| | |
| | | StoredState current = load(); |
| | | List<CabinetCandidate> cabinets = current.cabinets().stream() |
| | | .filter(item -> item.expiresAt().isAfter(now)).toList(); |
| | | List<CollectorCandidate> 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); |
| | | if (cabinets.size() != current.cabinets().size()) save(new StoredState(CURRENT_VERSION, cabinets)); |
| | | return new CandidateSnapshot(cabinets); |
| | | } |
| | | |
| | | private StoredState load() |
| | | { |
| | | Object stored = redisTemplate.opsForValue().get(STATE_KEY); |
| | | if (stored == null) return new StoredState(CURRENT_VERSION, List.of(), List.of()); |
| | | if (stored == null) return new StoredState(CURRENT_VERSION, List.of()); |
| | | try |
| | | { |
| | | // 通用 RedisTemplate 可能把泛型数组恢复为 JSONObject;在存储边界逐字段重建,禁止业务层强转。 |
| | |
| | | if (version == null || version != CURRENT_VERSION) |
| | | throw new IllegalStateException("Unsupported device discovery state version: " + version); |
| | | List<CabinetCandidate> cabinets = new ArrayList<>(); |
| | | // collectors 是模拟器时代写进同一 key 的历史字段:读到就忽略,不再恢复也不回写。 |
| | | requiredArray(root, "cabinets").forEach(item -> cabinets.add(cabinetCandidate(item))); |
| | | List<CollectorCandidate> collectors = new ArrayList<>(); |
| | | requiredArray(root, "collectors").forEach(item -> collectors.add(collectorCandidate(item))); |
| | | return new StoredState(CURRENT_VERSION, cabinets, collectors); |
| | | return new StoredState(CURRENT_VERSION, cabinets); |
| | | } |
| | | catch (IllegalStateException exception) |
| | | { |
| | |
| | | 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")); |
| | | } |
| | | |
| | | 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")); |
| | | latestSnapshot, optionalInstant(object, "lastDataAt"), object.getString("sourceDisplayNo")); |
| | | } |
| | | |
| | | private Instant requiredInstant(JSONObject object, String field) |
| | |
| | | redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(state)); |
| | | } |
| | | |
| | | private record StoredState(int version, List<CabinetCandidate> cabinets, |
| | | List<CollectorCandidate> collectors) |
| | | private record StoredState(int version, List<CabinetCandidate> cabinets) |
| | | { |
| | | private StoredState |
| | | { |
| | | cabinets = List.copyOf(cabinets); |
| | | collectors = List.copyOf(collectors); |
| | | } |
| | | } |
| | | } |