完成设备发现登记与状态事实

yeg

2026-09-05 bc9a169ce3648467136a279d167f72c4fbab9595

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/DeviceAdapterServiceImpl.java

@@ -3,30 +3,28 @@
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
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.ruoyi.device.api.DeviceAdapterException;
import com.ruoyi.device.model.AdapterCapabilities;
import com.ruoyi.device.model.ChannelIdentity;
import com.ruoyi.device.model.DetectedChannel;
import com.ruoyi.device.model.RealtimeSnapshot;
import com.ruoyi.device.registry.AdapterInstanceConfig;
import com.ruoyi.device.registry.AdapterRegistryException;
import com.ruoyi.device.registry.DeviceAdapterRegistry;
import com.battery.device.api.DeviceAdapterException;
import com.battery.device.simulator.BtsDevelopmentSimulator;
import com.battery.device.discovery.DeviceDiscoveryCandidateStore;
import com.battery.device.model.AdapterCapabilities;
import com.battery.device.registry.AdapterInstanceConfig;
import com.battery.device.registry.AdapterRegistryException;
import com.battery.device.registry.DeviceAdapterRegistry;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TDeviceAdapter;
import com.ruoyi.system.domain.vo.AdapterDiscoveryResult;
import com.ruoyi.system.mapper.TDeviceAdapterMapper;
import com.ruoyi.system.service.DeviceAdapterErrorCodes;
import com.ruoyi.system.service.IDeviceAdapterService;
import com.ruoyi.system.service.impl.DeviceDiscoveryPersistenceService.DiscoverySource;

@Service
public class DeviceAdapterServiceImpl implements IDeviceAdapterService
@@ -39,26 +37,23 @@
    private final TDeviceAdapterMapper mapper;
    private final DeviceAdapterRegistry registry;
    private final DeviceDiscoveryPersistenceService persistence;
    private final DeviceDiscoveryCandidateStore candidateStore;

    public DeviceAdapterServiceImpl(TDeviceAdapterMapper mapper, DeviceAdapterRegistry registry,
            DeviceDiscoveryPersistenceService persistence)
            DeviceDiscoveryPersistenceService persistence, DeviceDiscoveryCandidateStore candidateStore)
    {
        this.mapper = mapper;
        this.registry = registry;
        this.persistence = persistence;
        this.candidateStore = candidateStore;
    }

    DeviceAdapterServiceImpl(TDeviceAdapterMapper mapper, DeviceAdapterRegistry registry)
    {
        this(mapper, registry, new DeviceDiscoveryPersistenceService(mapper));
    }

    @Override
    /** 按管理端筛选条件查询适配器及其运行状态。 */
    @Override
    public List<TDeviceAdapter> selectList(TDeviceAdapter query) { return mapper.selectDeviceAdapterList(query); }

    @Override
    /** 查询适配器详情；实例不存在时抛出稳定业务错误。 */
    @Override
    public TDeviceAdapter selectById(Long adapterId)
    {
        TDeviceAdapter adapter = mapper.selectDeviceAdapterById(adapterId);
@@ -66,29 +61,48 @@
        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("DISCONNECTED");
        adapter.setConnectionStatus("OFFLINE");
        adapter.setDelFlag("0");
        try { return mapper.insertDeviceAdapter(adapter); }
        catch (DuplicateKeyException exception) { throw error("适配器编码已存在", DeviceAdapterErrorCodes.DUPLICATE_CODE); }
    }

    @Override
    /** 更新配置；已启用实例会同步运行时注册表。 */
    @Override
    public int update(TDeviceAdapter adapter)
    {
        TDeviceAdapter current = selectById(adapter.getAdapterId());
        normalizeAndValidate(adapter);
        adapter.setAdapterCode(current.getAdapterCode());
        adapter.setCapabilities(READ_ONLY_CAPABILITIES);
        adapter.setCapabilities(current.getCapabilities());
        int rows = mapper.updateDeviceAdapter(adapter);
        if (ENABLED.equals(current.getStatus()))
        {
@@ -98,8 +112,8 @@
        return rows;
    }

    @Override
    /** 启停适配器并同步其运行时实例。 */
    @Override
    public int changeStatus(Long adapterId, boolean enabled, String updateBy)
    {
        TDeviceAdapter adapter = selectById(adapterId);
@@ -111,8 +125,8 @@
        return rows;
    }

    /** 执行一次发现和实时观测，将未登记设备写入 Redis 候选。 */
    @Override
    /** 执行一次发现和实时观测，并事务性写入通道事实。 */
    public AdapterDiscoveryResult discover(Long adapterId)
    {
        TDeviceAdapter adapter = selectById(adapterId);
@@ -121,47 +135,57 @@
        try
        {
            registry.synchronize(toRuntimeConfig(adapter));
            DiscoveryObservation observation = registry.execute(adapter.getAdapterCode(), runtime -> {
                List<DetectedChannel> channels = runtime.discoverChannels();
                Set<ChannelIdentity> available = channels.stream()
                        .filter(DetectedChannel::available)
                        .map(DetectedChannel::identity)
                        .collect(Collectors.toCollection(LinkedHashSet::new));
                return new DiscoveryObservation(channels, runtime.inquire(available));
            });
            Instant observedAt = Instant.now();
            AdapterDiscoveryResult result = persistence.persist(adapter, observation.channels(),
                    observation.snapshots(), observedAt);
            mapper.updateDiscoverySuccess(adapterId, observedAt);
            AdapterDiscoveryResult result = persistence.discoverForReception(registry, new DiscoverySource(
                    adapter.getAdapterCode(), adapter.getAdapterId(), adapter.getAdapterName(),
                    adapter.getPollIntervalMs()));
            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)
    {
        selectById(adapterId);
        return mapper.selectDetectionChannelsByAdapterId(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
    /** 启动时恢复所有已启用适配器的运行时注册。 */
    @Override
    public void synchronizeEnabledAdapters()
    {
        mapper.selectEnabledDeviceAdapters().forEach(adapter -> registry.synchronize(toRuntimeConfig(adapter)));
    }

    @Override
    /** 轮询达到周期的已启用适配器，并持久化失败状态后继续调度。 */
    @Override
    public void pollDueAdapters()
    {
        Instant now = Instant.now();
@@ -169,7 +193,11 @@
        {
            Instant lastAttempt = adapter.getLastAttemptAtUtc();
            if (lastAttempt != null && Duration.between(lastAttempt, now).toMillis() < adapter.getPollIntervalMs()) continue;
            try { discover(adapter.getAdapterId()); }
            try
            {
                // 设备发现必须持续执行；暂停接收只由 discoverNow 中的 receivingIdentities 控制实时快照请求。
                discover(adapter.getAdapterId());
            }
            catch (ServiceException ignored) { /* 失败状态已持久化，下一轮按周期重试。 */ }
        }
    }
@@ -206,17 +234,27 @@
        }
    }

    private void rejectDevelopmentSimulatorDatabaseConfig(TDeviceAdapter adapter)
    {
        if (BtsDevelopmentSimulator.ADAPTER_CODE.equals(adapter.getAdapterCode()))
            throw error("dev-sim 由开发环境代码固定装配，不能写入数据库", 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()), AdapterCapabilities.observeOnly());
                ENABLED.equals(adapter.getStatus()), parseCapabilities(adapter.getCapabilities()));
    }

    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); }

    private record DiscoveryObservation(List<DetectedChannel> channels,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots)
    {
    }
}