package com.battery.business.service.impl; import java.time.Duration; import java.time.Instant; import java.time.ZoneId; import java.util.List; import java.util.Locale; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.dao.DuplicateKeyException; import org.springframework.stereotype.Service; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.ruoyi.common.exception.ServiceException; import com.ruoyi.common.utils.StringUtils; import com.battery.device.api.DeviceAdapterException; import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore; import com.battery.device.api.model.AdapterCapabilities; import com.battery.device.api.registry.AdapterInstanceConfig; import com.battery.device.api.registry.AdapterRegistryException; import com.battery.device.api.registry.DeviceAdapterRegistry; import com.battery.business.domain.TDetectionChannel; import com.battery.business.domain.TDeviceAdapter; import com.battery.business.domain.vo.AdapterDiscoveryResult; import com.battery.business.mapper.TDeviceAdapterMapper; import com.battery.business.service.DeviceAdapterErrorCodes; import com.battery.business.service.IDeviceAdapterService; import com.battery.business.service.impl.DeviceDiscoveryPersistenceService.DiscoverySource; @Service public class DeviceAdapterServiceImpl implements IDeviceAdapterService { private static final Logger log = LoggerFactory.getLogger(DeviceAdapterServiceImpl.class); private static final String ENABLED = "0"; private static final String DISABLED = "1"; private static final String READ_ONLY_CAPABILITIES = "{\"discovery\":true,\"realtimeObservation\":true,\"historyDownload\":true,\"control\":false}"; /** * 适配器类型按存储形态划分;BTS 保留给开发模拟器与将来的协议路线(ADR-0065)。 * `NEWARE_MYSQL` 只能写字面量:battery-business 不依赖 battery-device-neware, * 拿不到那边的常量;推式来源的实现就在本模块,所以用它的常量。 */ private static final List SUPPORTED_TYPES = List.of("BTS", "NEWARE_MYSQL", PushIngestAdapter.ADAPTER_TYPE); /** * 通道清单长期存在的来源,候选不按时间过期:直连厂商存储的来源读的是厂商的登记表, * 推式来源读的是取数程序上报的拓扑——两者都不是"轮询发现"的结果。 */ private static final List PERSISTENT_CANDIDATE_TYPES = List.of("NEWARE_MYSQL", PushIngestAdapter.ADAPTER_TYPE); private final TDeviceAdapterMapper mapper; private final DeviceAdapterRegistry registry; private final DeviceDiscoveryPersistenceService persistence; private final DeviceDiscoveryCandidateStore candidateStore; public DeviceAdapterServiceImpl(TDeviceAdapterMapper mapper, DeviceAdapterRegistry registry, DeviceDiscoveryPersistenceService persistence, DeviceDiscoveryCandidateStore candidateStore) { this.mapper = mapper; this.registry = registry; this.persistence = persistence; this.candidateStore = candidateStore; } /** 按管理端筛选条件查询适配器及其运行状态。 */ @Override public List selectList(TDeviceAdapter query) { return mapper.selectDeviceAdapterList(query); } /** 查询适配器详情;实例不存在时抛出稳定业务错误。 */ @Override public TDeviceAdapter selectById(Long adapterId) { TDeviceAdapter adapter = mapper.selectDeviceAdapterById(adapterId); if (adapter == null) throw error("设备适配器实例不存在", DeviceAdapterErrorCodes.NOT_FOUND); return adapter; } /** 幂等注册应用托管的只读适配器;不预写检测柜或通道事实。 */ @Override public TDeviceAdapter ensureManagedAdapter(TDeviceAdapter adapter) { normalizeAndValidate(adapter); rejectDevelopmentSimulatorDatabaseConfig(adapter); AdapterCapabilities capabilities = parseCapabilities(adapter.getCapabilities()); if (capabilities.control()) throw error("当前里程碑只允许只读设备适配器", DeviceAdapterErrorCodes.INVALID_CONFIG); adapter.setStatus(ENABLED); adapter.setConnectionStatus("OFFLINE"); adapter.setDelFlag("0"); mapper.upsertManagedDeviceAdapter(adapter); TDeviceAdapter registered = mapper.selectDeviceAdapterByCode(adapter.getAdapterCode()); if (registered == null) throw error("托管设备适配器注册失败", DeviceAdapterErrorCodes.NOT_FOUND); return registered; } /** 新增默认禁用的只读适配器配置。 */ @Override public int insert(TDeviceAdapter adapter) { normalizeAndValidate(adapter); rejectDevelopmentSimulatorDatabaseConfig(adapter); if (mapper.selectDeviceAdapterByCode(adapter.getAdapterCode()) != null) throw error("适配器编码已存在", DeviceAdapterErrorCodes.DUPLICATE_CODE); adapter.setCapabilities(READ_ONLY_CAPABILITIES); adapter.setStatus(DISABLED); adapter.setConnectionStatus("OFFLINE"); adapter.setDelFlag("0"); try { return mapper.insertDeviceAdapter(adapter); } catch (DuplicateKeyException exception) { throw error("适配器编码已存在", DeviceAdapterErrorCodes.DUPLICATE_CODE); } } /** 更新配置;已启用实例会同步运行时注册表。 */ @Override public int update(TDeviceAdapter adapter) { TDeviceAdapter current = selectById(adapter.getAdapterId()); normalizeAndValidate(adapter); adapter.setAdapterCode(current.getAdapterCode()); adapter.setCapabilities(current.getCapabilities()); int rows = mapper.updateDeviceAdapter(adapter); if (ENABLED.equals(current.getStatus())) { adapter.setStatus(ENABLED); registry.synchronize(toRuntimeConfig(adapter)); } return rows; } /** 启停适配器并同步其运行时实例。 */ @Override public int changeStatus(Long adapterId, boolean enabled, String updateBy) { TDeviceAdapter adapter = selectById(adapterId); String status = enabled ? ENABLED : DISABLED; if (status.equals(adapter.getStatus())) return 1; adapter.setStatus(status); // 先装配运行时实例:装配失败(例如缺少口令环境变量)时不把状态写成"已启用", // 免得留下"库里启用、运行时没有实例"的假状态。 registry.synchronize(toRuntimeConfig(adapter)); return mapper.updateDeviceAdapterStatus(adapterId, status, updateBy); } /** 执行一次发现和实时观测,将未登记设备写入 Redis 候选。 */ @Override public AdapterDiscoveryResult discover(Long adapterId) { TDeviceAdapter adapter = selectById(adapterId); if (!ENABLED.equals(adapter.getStatus())) throw error("设备适配器实例未启用", DeviceAdapterErrorCodes.DISABLED); try { registry.synchronize(toRuntimeConfig(adapter)); AdapterDiscoveryResult result = persistence.discoverForReception(registry, new DiscoverySource( adapter.getAdapterCode(), adapter.getAdapterId(), adapter.getAdapterName(), adapter.getPollIntervalMs(), PERSISTENT_CANDIDATE_TYPES.contains(adapter.getAdapterType()))); mapper.updateDiscoverySuccess(adapterId, result.observedAtUtc()); return result; } catch (DeviceAdapterException exception) { persistence.markUnavailable(adapter.getAdapterCode()); return failDiscovery(adapterId, exception.failureCode(), exception); } catch (AdapterRegistryException exception) { persistence.markUnavailable(adapter.getAdapterCode()); return failDiscovery(adapterId, "ADAPTER_FAILURE", exception); } } /** 查询指定适配器当前未过期的 Redis 发现通道。 */ @Override public List selectChannels(Long adapterId) { TDeviceAdapter adapter = selectById(adapterId); return candidateStore.snapshot(Instant.now()).cabinets().stream() .filter(item -> item.adapterCode().equals(adapter.getAdapterCode())) .flatMap(item -> item.channels().stream().map(channel -> { TDetectionChannel value = new TDetectionChannel(); value.setAdapterId(adapterId); value.setDevtype(channel.identity().devtype()); value.setDevid(channel.identity().devid()); value.setSubdevid(channel.identity().subdevid()); value.setChlid(channel.identity().chlid()); value.setDisplayNo(channel.identity().subdevid() + "-" + channel.identity().chlid()); value.setOnlineStatus(channel.online() ? "ONLINE" : "OFFLINE"); value.setWorkStatus(channel.workStatus().name()); value.setLastSeenAtUtc(channel.observedAt()); value.setStatus("0"); return value; })) .toList(); } /** 启动时恢复所有已启用适配器的运行时注册。 */ @Override public void synchronizeEnabledAdapters() { for (TDeviceAdapter adapter : mapper.selectEnabledDeviceAdapters()) { try { registry.synchronize(toRuntimeConfig(adapter)); } catch (RuntimeException exception) { // 单行配置有问题(例如没放凭据文件、口令键名写错)不得拦住后端启动: // 既写日志,也按发现失败的同一路径落到这一行,使"忘了配"在启动后立刻是列表里可见的一行。 log.warn("设备适配器实例装配失败 adapterCode={}:{}", adapter.getAdapterCode(), exception.getMessage()); mapper.updateDiscoveryFailure(adapter.getAdapterId(), startupFailureCode(exception), exception.getMessage()); } } } /** * 装配失败写进实例行的错误码:适配器自己给出分类(例如凭据缺失、连接失败)时用它, * 否则记为配置问题——启动阶段的装配失败基本都是配置造成的。 */ private static String startupFailureCode(RuntimeException exception) { return exception instanceof DeviceAdapterException deviceFailure && !deviceFailure.failureCode().isBlank() ? deviceFailure.failureCode() : "INVALID_CONFIG"; } /** 轮询达到周期的已启用适配器,并持久化失败状态后继续调度。 */ @Override public void pollDueAdapters() { Instant now = Instant.now(); for (TDeviceAdapter adapter : mapper.selectEnabledDeviceAdapters()) { Instant lastAttempt = adapter.getLastAttemptAtUtc(); if (lastAttempt != null && Duration.between(lastAttempt, now).toMillis() < adapter.getPollIntervalMs()) continue; try { // 设备发现必须持续执行;暂停接收只由 discoverNow 中的 receivingIdentities 控制实时快照请求。 discover(adapter.getAdapterId()); } catch (ServiceException ignored) { /* 失败状态已持久化,下一轮按周期重试。 */ } } } private AdapterDiscoveryResult failDiscovery(Long adapterId, String errorCode, RuntimeException exception) { mapper.updateDiscoveryFailure(adapterId, errorCode, exception.getMessage()); throw error("设备发现失败:" + exception.getMessage(), DeviceAdapterErrorCodes.DISCOVERY_FAILED); } private void normalizeAndValidate(TDeviceAdapter adapter) { adapter.setAdapterCode(StringUtils.trim(adapter.getAdapterCode())); adapter.setAdapterName(StringUtils.trim(adapter.getAdapterName())); adapter.setAdapterType(StringUtils.trim(adapter.getAdapterType()).toUpperCase(Locale.ROOT)); adapter.setHost(StringUtils.trim(adapter.getHost())); adapter.setTimezoneId(StringUtils.trim(adapter.getTimezoneId())); adapter.setSourceConfig(StringUtils.trim(adapter.getSourceConfig())); adapter.setRemark(StringUtils.trim(adapter.getRemark())); if (!SUPPORTED_TYPES.contains(adapter.getAdapterType())) throw error("不支持的适配器类型", DeviceAdapterErrorCodes.UNSUPPORTED_TYPE); if (!Boolean.TRUE.equals(adapter.getReadOnly())) throw error("当前里程碑只允许只读设备适配器", DeviceAdapterErrorCodes.INVALID_CONFIG); try { if (adapter.getPort() == null || adapter.getPort() < 1 || adapter.getPort() > 65535 || adapter.getPollIntervalMs() == null || adapter.getPollIntervalMs() < 1000 || StringUtils.isBlank(adapter.getAdapterCode()) || StringUtils.isBlank(adapter.getAdapterName()) || StringUtils.isBlank(adapter.getHost())) throw new IllegalArgumentException(); ZoneId.of(adapter.getTimezoneId()); } catch (RuntimeException exception) { throw error("设备适配器配置不正确", DeviceAdapterErrorCodes.INVALID_CONFIG); } } private void rejectDevelopmentSimulatorDatabaseConfig(TDeviceAdapter adapter) { if (registry.contains(adapter.getAdapterCode()) && mapper.selectDeviceAdapterByCode(adapter.getAdapterCode()) == null) throw error("该适配器由运行环境装配,不能写入数据库", DeviceAdapterErrorCodes.INVALID_CONFIG); } private AdapterInstanceConfig toRuntimeConfig(TDeviceAdapter adapter) { return new AdapterInstanceConfig(adapter.getAdapterCode(), adapter.getAdapterType(), adapter.getHost(), adapter.getPort(), adapter.getPollIntervalMs(), adapter.getTimezoneId(), ENABLED.equals(adapter.getStatus()), parseCapabilities(adapter.getCapabilities()), parseSourceConfig(adapter.getSourceConfig())); } /** 来源形态配置只承载非敏感项;连接口令由环境变量或启动参数提供,不入库。 */ private static java.util.Map parseSourceConfig(String sourceConfig) { if (StringUtils.isBlank(sourceConfig)) return java.util.Map.of(); JSONObject value = JSON.parseObject(sourceConfig); if (value == null) return java.util.Map.of(); java.util.Map result = new java.util.LinkedHashMap<>(); value.forEach((key, item) -> { if (item != null) result.put(String.valueOf(key), String.valueOf(item)); }); return result; } private AdapterCapabilities parseCapabilities(String capabilities) { JSONObject value = JSON.parseObject(capabilities); if (value == null) throw error("设备适配器能力配置不正确", DeviceAdapterErrorCodes.INVALID_CONFIG); return new AdapterCapabilities(value.getBooleanValue("discovery"), value.getBooleanValue("realtimeObservation"), value.getBooleanValue("historyDownload"), value.getBooleanValue("control")); } private ServiceException error(String message, int code) { return new ServiceException(message, code); } }