| | |
| | | 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; |
| | |
| | | this.persistence = persistence; |
| | | } |
| | | |
| | | 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); |
| | |
| | | return adapter; |
| | | } |
| | | |
| | | /** 幂等注册应用托管的只读适配器;不预写检测柜或通道事实。 */ |
| | | @Override |
| | | public TDeviceAdapter ensureManagedAdapter(TDeviceAdapter adapter) |
| | | { |
| | | normalizeAndValidate(adapter); |
| | | AdapterCapabilities capabilities = parseCapabilities(adapter.getCapabilities()); |
| | | if (capabilities.control()) |
| | | throw error("当前里程碑只允许只读设备适配器", DeviceAdapterErrorCodes.INVALID_CONFIG); |
| | | adapter.setStatus(ENABLED); |
| | | adapter.setConnectionStatus("DISCONNECTED"); |
| | | 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); |
| | |
| | | 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())) |
| | | { |
| | |
| | | return rows; |
| | | } |
| | | |
| | | @Override |
| | | /** 启停适配器并同步其运行时实例。 */ |
| | | @Override |
| | | public int changeStatus(Long adapterId, boolean enabled, String updateBy) |
| | | { |
| | | TDeviceAdapter adapter = selectById(adapterId); |
| | |
| | | return rows; |
| | | } |
| | | |
| | | @Override |
| | | /** 执行一次发现和实时观测,并事务性写入通道事实。 */ |
| | | @Override |
| | | public AdapterDiscoveryResult discover(Long adapterId) |
| | | { |
| | | TDeviceAdapter adapter = selectById(adapterId); |
| | |
| | | .filter(DetectedChannel::available) |
| | | .map(DetectedChannel::identity) |
| | | .collect(Collectors.toCollection(LinkedHashSet::new)); |
| | | return new DiscoveryObservation(channels, runtime.inquire(available)); |
| | | return new DiscoveryObservation(channels, available.isEmpty() ? Map.of() : runtime.inquire(available)); |
| | | }); |
| | | Instant observedAt = Instant.now(); |
| | | AdapterDiscoveryResult result = persistence.persist(adapter, observation.channels(), |
| | |
| | | } |
| | | } |
| | | |
| | | @Override |
| | | /** 查询指定适配器已发现且未删除的通道。 */ |
| | | @Override |
| | | public List<TDetectionChannel> selectChannels(Long adapterId) |
| | | { |
| | | selectById(adapterId); |
| | | return mapper.selectDetectionChannelsByAdapterId(adapterId); |
| | | } |
| | | |
| | | @Override |
| | | /** 启动时恢复所有已启用适配器的运行时注册。 */ |
| | | @Override |
| | | public void synchronizeEnabledAdapters() |
| | | { |
| | | mapper.selectEnabledDeviceAdapters().forEach(adapter -> registry.synchronize(toRuntimeConfig(adapter))); |
| | | } |
| | | |
| | | @Override |
| | | /** 轮询达到周期的已启用适配器,并持久化失败状态后继续调度。 */ |
| | | @Override |
| | | public void pollDueAdapters() |
| | | { |
| | | Instant now = Instant.now(); |
| | |
| | | { |
| | | 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); } |