package com.battery.business.service.impl; import java.util.List; import java.util.Map; import java.util.Set; import org.springframework.context.annotation.Primary; import org.springframework.stereotype.Service; import com.battery.device.api.TestDataObservationAdapter; import com.battery.device.api.model.AdapterCapabilities; import com.battery.device.api.model.ChannelIdentity; import com.battery.device.api.model.DetectedChannel; import com.battery.device.api.model.DeviceObservation; import com.battery.device.api.registry.AdapterRegistryException; import com.battery.device.api.registry.DeviceAdapterRegistry; /** * 把取数请求按通道身份里的适配器编码送到它所属的来源实例。 * *

这是持久化与实时展示路径上唯一的取数入口:同一个进程里可以注册多个来源实例, * 每个通道的请求只送到它自己的实例;实例缺失或未提供测试数据观测契约时明确失败, * 不回退到其他实例。发现不走这里,它按实例在 {@link DeviceDiscoveryPersistenceService} 中执行。 * *

声明为 {@code @Primary}:来源实例可能自己也实现了观测契约(开发模拟器就是这样, * 它同时是 Spring Bean),业务侧注入的必须是这个按编码解析的入口,而不是某台设备的实例。 */ @Service @Primary public class RoutingTestDataObservationAdapter implements TestDataObservationAdapter { /** 聚合路由不是来源实例,这个编码只用于说明"按通道身份路由"。 */ public static final String ADAPTER_CODE = "by-source"; private final DeviceAdapterRegistry registry; public RoutingTestDataObservationAdapter(DeviceAdapterRegistry registry) { this.registry = registry; } /** 返回聚合路由的固定编码,仅用于说明"按通道身份路由",不是任何来源实例的编码。 */ @Override public String adapterCode() { return ADAPTER_CODE; } /** 返回聚合适配器声明的能力:只观测,不提供控制能力。 */ @Override public AdapterCapabilities capabilities() { return AdapterCapabilities.observeOnly(); } /** 聚合适配器不承载发现;发现按来源实例执行,避免在这条路上猜一个设备。 */ @Override public List discoverChannels() { throw new UnsupportedOperationException("设备发现按来源实例执行,不在聚合适配器上执行"); } /** 一次观测里的通道可能来自多个来源实例,逐个按各自的编码请求,不能按第一个通道决定全部。 */ @Override public Map observe(Set channels) { if (channels == null || channels.isEmpty()) return Map.of(); Map result = new java.util.LinkedHashMap<>(); for (Map.Entry> group : groupBySource(channels).entrySet()) { Set groupChannels = new java.util.LinkedHashSet<>(group.getValue()); result.putAll(registry.executeTestData(group.getKey(), adapter -> adapter.observe(groupChannels))); } return Map.copyOf(result); } private static Map> groupBySource(Set channels) { Map> grouped = new java.util.LinkedHashMap<>(); for (ChannelIdentity channel : channels) grouped.computeIfAbsent(channel.adapterCode(), ignored -> new java.util.ArrayList<>()).add(channel); return grouped; } /** * 按通道所属来源实例请求该通道最近一次实时观测。 * * @param channel 检测柜通道身份,其中的适配器编码决定请求送到哪个实例 * @return 最近一次实时观测;没有成功观测时为空 * @throws AdapterRegistryException 该通道所属的实例未启用,或实例未提供测试数据观测契约 */ @Override public CabinetRealtime latestCabinet(ChannelIdentity channel) { return registry.executeTestData(channel.adapterCode(), adapter -> adapter.latestCabinet(channel)); } /** * 按来源实例请求采集器的辅助测量快照。 * * @param adapterCode 来源实例编码;为空表示来源未知,直接返回空值而不猜也不回退 * @param collectorCode 采集器编码 * @return 采集器实时快照;来源未知时为空 * @throws AdapterRegistryException 来源实例未启用,或实例未提供测试数据观测契约 */ @Override public CollectorRealtime latestCollector(String adapterCode, String collectorCode) { // 采集器测量属于某个来源实例的辅助测量,按该实例请求;实例缺失时明确失败。 // 来源编码为空说明测试电池连检测柜绑定都没有,此时来源未知:不猜也不回退,报无法请求。 if (adapterCode == null || adapterCode.isBlank()) return null; return registry.executeTestData(adapterCode, adapter -> adapter.latestCollector(adapterCode, collectorCode)); } /** * 按通道所属来源实例请求设备直接提供的工步统计。 * * @param channel 检测柜通道身份 * @return 该通道的设备工步统计;设备未提供时为空列表 * @throws AdapterRegistryException 该通道所属的实例未启用,或实例未提供测试数据观测契约 */ @Override public List stepRecords(ChannelIdentity channel) { return registry.executeTestData(channel.adapterCode(), adapter -> adapter.stepRecords(channel)); } /** * 按通道所属来源实例请求循环统计,来源类别由实例写在记录上。 * * @param channel 检测柜通道身份 * @return 该通道的循环统计;设备未提供且无法聚合时为空列表 * @throws AdapterRegistryException 该通道所属的实例未启用,或实例未提供测试数据观测契约 */ @Override public List cycleRecords(ChannelIdentity channel) { return registry.executeTestData(channel.adapterCode(), adapter -> adapter.cycleRecords(channel)); } /** * 按通道所属来源实例取该来源定义的实时持久化槽位间隔。 * * @param channel 检测柜通道身份 * @return 持久化槽位间隔,单位毫秒 * @throws AdapterRegistryException 该通道所属的实例未启用,或实例未提供测试数据观测契约 */ @Override public long persistenceIntervalMillis(ChannelIdentity channel) { return registry.executeTestData(channel.adapterCode(), adapter -> adapter.persistenceIntervalMillis(channel)); } /** * 按通道所属来源实例取工步与循环的读取间隔。 * *

必须在这里转交:不转交就会落到本类的默认实现,从而拿实时持久化间隔去重算汇总, * 来源自己声明的汇总节奏就被忽略了。 * * @param channel 检测柜通道身份 * @return 汇总记录的读取间隔,单位毫秒 * @throws AdapterRegistryException 该通道所属的实例未启用,或实例未提供测试数据观测契约 */ @Override public long aggregateIntervalMillis(ChannelIdentity channel) { return registry.executeTestData(channel.adapterCode(), adapter -> adapter.aggregateIntervalMillis(channel)); } /** * 按通道所属来源实例取设备计划方案,只用于实时展示。 * * @param channel 检测柜通道身份 * @return 设备计划方案;设备未提供时为空 * @throws AdapterRegistryException 该通道所属的实例未启用,或实例未提供测试数据观测契约 */ @Override public TestPlan testPlan(ChannelIdentity channel) { return registry.executeTestData(channel.adapterCode(), adapter -> adapter.testPlan(channel)); } /** * 按通道所属来源实例读取接收水位,用于把开始接收与继续接收切在这一时刻之后。 * * @param channel 检测柜通道身份 * @return 通道当前接收水位;来源不按设备数据行接收时为空 * @throws AdapterRegistryException 该通道所属的实例未启用,或实例未提供测试数据观测契约 */ @Override public SampleWatermark currentWatermark(ChannelIdentity channel) { return registry.executeTestData(channel.adapterCode(), adapter -> adapter.currentWatermark(channel)); } /** * 按通道所属来源实例读取严格晚于水位的数据行,同秒多行各自成立、不按秒合并。 * * @param channel 检测柜通道身份 * @param watermark 上次取得的水位;为空表示尚未取水位,此时返回空列表而不并入设备历史 * @return 水位之后的设备数据行,按设备时间与协议序号升序 * @throws AdapterRegistryException 该通道所属的实例未启用,或实例未提供测试数据观测契约 */ @Override public List samplesSince(ChannelIdentity channel, SampleWatermark watermark) { return registry.executeTestData(channel.adapterCode(), adapter -> adapter.samplesSince(channel, watermark)); } }