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<Object, Object> redisTemplate;
|
|
public RedisDeviceDiscoveryCandidateStore(RedisTemplate<Object, Object> redisTemplate)
|
{
|
this.redisTemplate = redisTemplate;
|
}
|
|
/** 按设备编码 upsert 本轮检测柜候选，本轮未返回的候选保留至逻辑过期。 */
|
@Override
|
public synchronized void recordCabinets(List<CabinetCandidate> candidates)
|
{
|
StoredState current = load();
|
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()));
|
}
|
|
/** 替换指定设备适配器实例的全部采集器候选，其它实例的候选保持不变。 */
|
@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));
|
}
|
|
/**
|
* 返回指定时刻尚未过期的发现候选，并顺带清掉已逻辑过期的条目。
|
*
|
* <p>是否“在用”由操作员判断，适配器不自动判定。</p>
|
*/
|
@Override
|
public synchronized CandidateSnapshot snapshot(Instant now)
|
{
|
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);
|
}
|
|
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<CabinetCandidate> cabinets = new ArrayList<>();
|
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);
|
}
|
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<ChannelCandidate> 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);
|
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"));
|
}
|
|
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<CabinetCandidate> cabinets,
|
List<CollectorCandidate> collectors)
|
{
|
private StoredState
|
{
|
cabinets = List.copyOf(cabinets);
|
collectors = List.copyOf(collectors);
|
}
|
}
|
}
|
