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<String> SUPPORTED_TYPES = List.of("BTS", "NEWARE_MYSQL", PushIngestAdapter.ADAPTER_TYPE);
|
/**
|
* 通道清单长期存在的来源，候选不按时间过期：直连厂商存储的来源读的是厂商的登记表，
|
* 推式来源读的是取数程序上报的拓扑——两者都不是"轮询发现"的结果。
|
*/
|
private static final List<String> 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<TDeviceAdapter> 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<TDetectionChannel> 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<String, String> 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<String, String> 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); }
|
}
|
