From 147428971be43eed90a79f3849f15000d3855390 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期一, 28 九月 2026 15:15:22 +0800
Subject: [PATCH] 实现新威 7.6 直连取数与逐行接收
---
battery-business/src/main/java/com/battery/business/service/impl/TDetectionCabinetServiceImpl.java | 51
battery-business/src/main/java/com/battery/business/domain/vo/DiscoveredCabinetVo.java | 12
battery-business/src/main/resources/mapper/battery/TTestMapper.xml | 6
battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java | 130 ++
battery-device-neware/src/test/java/com/battery/device/neware/FakeVendorDatabase.java | 333 +++++
ruoyi-admin/src/test/java/com/battery/business/service/impl/VendorSampleReceptionServiceTest.java | 261 ++++
battery-device-neware/src/main/java/com/battery/device/neware/NonQueryStatementException.java | 12
ruoyi-admin/src/main/resources/db/migration-final/V22__vendor_source_instance_config.sql | 3
battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java | 140 ++
ruoyi-admin/src/main/java/com/battery/business/runtime/DeviceAdapterRuntimeConfig.java | 22
battery-device-neware/src/test/java/com/battery/device/neware/ReadOnlyVendorDataSourceTest.java | 161 ++
battery-device-api/src/main/java/com/battery/device/api/registry/DeviceAdapterRegistry.java | 75 +
battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyStatementAudit.java | 45
battery-business/src/main/java/com/battery/business/domain/TDetectionCabinet.java | 7
ruoyi-admin/src/test/java/com/battery/business/runtime/VendorSourceInstanceMigrationContractTest.java | 97 +
ruoyi-admin/src/test/java/com/battery/business/service/impl/TestDataReceptionServiceImplTest.java | 14
battery-business/src/main/java/com/battery/business/mapper/TTestMapper.java | 3
battery-device-api/src/main/java/com/battery/device/api/discovery/DeviceDiscoveryCandidateStore.java | 27
ruoyi-admin/src/main/java/com/battery/business/runtime/RedisDeviceDiscoveryCandidateStore.java | 12
battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterRuntimeCoordinator.java | 7
ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentSimulatorConfig.java | 4
ruoyi-admin/src/test/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceServiceTest.java | 36
ruoyi-admin/src/test/java/com/battery/business/runtime/DeviceAdapterRuntimeConfigTest.java | 9
battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapterFactory.java | 78 +
battery-device-neware/src/test/java/com/battery/device/neware/NewareMysqlAdapterContractTest.java | 207 +++
.gitignore | 1
battery-business/src/main/resources/mapper/battery/TDeviceAdapterMapper.xml | 15
battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java | 47
battery-device-api/src/main/java/com/battery/device/api/registry/AdapterInstanceConfig.java | 21
battery-device-neware/pom.xml | 13
battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java | 7
pom.xml | 6
battery-device-neware/src/main/java/com/battery/device/neware/SingleConnectionVendorDataSource.java | 110 +
battery-device-simulator/src/main/java/com/battery/device/simulator/BtsDevelopmentSimulator.java | 43
ruoyi-admin/src/main/java/com/battery/business/runtime/TestDataReceptionScheduler.java | 22
battery-device-simulator/src/test/java/com/battery/device/simulator/SimulatorObservationAdapterContractTest.java | 8
ruoyi-admin/src/test/java/com/battery/business/service/impl/TTestLiveServiceImplTest.java | 4
ruoyi-admin/pom.xml | 6
battery-business/src/main/java/com/battery/business/domain/TDeviceAdapter.java | 6
battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyVendorDataSource.java | 237 ++++
battery-device-api/src/main/java/com/battery/device/api/model/DetectedChannel.java | 17
ruoyi-admin/src/test/java/com/battery/business/web/VendorSamplePersistenceDatabaseTest.java | 156 ++
/dev/null | 19
ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingAdapterWiringContractTest.java | 33
battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapter.java | 466 ++++++++
battery-business/src/main/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceService.java | 28
battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterServiceImpl.java | 25
battery-device-api/src/test/java/com/battery/device/api/registry/DeviceAdapterRegistryTest.java | 135 ++
ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingTestDataObservationAdapterTest.java | 163 ++
49 files changed, 3,234 insertions(+), 106 deletions(-)
diff --git a/.gitignore b/.gitignore
index aa757e8..dff4da9 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,6 +4,7 @@
/battery-business/target/
/battery-device-api/target/
/battery-device-bts/target/
+/battery-device-neware/target/
/battery-device-simulator/target/
/ruoyi-framework/target/
/ruoyi-generator/target/
diff --git a/battery-business/src/main/java/com/battery/business/domain/TDetectionCabinet.java b/battery-business/src/main/java/com/battery/business/domain/TDetectionCabinet.java
index 48ca93e..f784ed2 100644
--- a/battery-business/src/main/java/com/battery/business/domain/TDetectionCabinet.java
+++ b/battery-business/src/main/java/com/battery/business/domain/TDetectionCabinet.java
@@ -30,6 +30,9 @@
private Long adapterId;
private String adapterCode;
+ /** 数据来源名称，仅供展示，用于区分不同机器上的同名柜号；不落库。 */
+ private String adapterName;
+
private Integer vendorDeviceType;
private Integer vendorDeviceId;
private Instant lastSeenAtUtc;
@@ -107,7 +110,9 @@
public Long getAdapterId() { return adapterId; }
public void setAdapterId(Long adapterId) { this.adapterId = adapterId; }
public String getAdapterCode() { return adapterCode; }
- public void setAdapterCode(String adapterCode) { this.adapterCode = adapterCode; }
+ public void setAdapterCode(String value) { this.adapterCode = value; }
+ public String getAdapterName() { return adapterName; }
+ public void setAdapterName(String value) { adapterName = value; }
public Integer getVendorDeviceType() { return vendorDeviceType; }
public void setVendorDeviceType(Integer value) { vendorDeviceType = value; }
public Integer getVendorDeviceId() { return vendorDeviceId; }
diff --git a/battery-business/src/main/java/com/battery/business/domain/TDeviceAdapter.java b/battery-business/src/main/java/com/battery/business/domain/TDeviceAdapter.java
index e5c38fa..ed700db 100644
--- a/battery-business/src/main/java/com/battery/business/domain/TDeviceAdapter.java
+++ b/battery-business/src/main/java/com/battery/business/domain/TDeviceAdapter.java
@@ -49,6 +49,10 @@
private Boolean readOnly;
private String capabilities;
+
+ /** 来源形态配置（JSON），承载各存储形态互不相交的非敏感字段；连接口令不入库。 */
+ private String sourceConfig;
+
private String status;
/** 适配器连接状态，仅允许 ONLINE/OFFLINE。 */
private String connectionStatus;
@@ -78,6 +82,8 @@
public void setReadOnly(Boolean readOnly) { this.readOnly = readOnly; }
public String getCapabilities() { return capabilities; }
public void setCapabilities(String capabilities) { this.capabilities = capabilities; }
+ public String getSourceConfig() { return sourceConfig; }
+ public void setSourceConfig(String sourceConfig) { this.sourceConfig = sourceConfig; }
public String getStatus() { return status; }
public void setStatus(String status) { this.status = status; }
public String getConnectionStatus() { return connectionStatus; }
diff --git a/battery-business/src/main/java/com/battery/business/domain/vo/DiscoveredCabinetVo.java b/battery-business/src/main/java/com/battery/business/domain/vo/DiscoveredCabinetVo.java
index 96d84de..ff8346e 100644
--- a/battery-business/src/main/java/com/battery/business/domain/vo/DiscoveredCabinetVo.java
+++ b/battery-business/src/main/java/com/battery/business/domain/vo/DiscoveredCabinetVo.java
@@ -1,6 +1,12 @@
package com.battery.business.domain.vo;
-/** 原检测柜选择弹窗使用的设备摘要。 */
+import java.time.Instant;
+
+/**
+ * 检测柜选择弹窗使用的设备摘要。
+ * {@code lastDataAt} 是该设备最近一次采样行的设备时间（来源里没有采样行时为空），
+ * 由操作员据此判断哪几台柜子现在真的在产出数据；登记与否仍由操作员决定。
+ */
public record DiscoveredCabinetVo(
String deviceCode,
String ip,
@@ -9,7 +15,8 @@
String manufacturer,
String model,
Integer channelCount,
- String onlineStatus)
+ String onlineStatus,
+ Instant lastDataAt)
{
public String getDeviceCode() { return deviceCode; }
public String getIp() { return ip; }
@@ -19,4 +26,5 @@
public String getModel() { return model; }
public Integer getChannelCount() { return channelCount; }
public String getOnlineStatus() { return onlineStatus; }
+ public Instant getLastDataAt() { return lastDataAt; }
}
diff --git a/battery-business/src/main/java/com/battery/business/mapper/TTestMapper.java b/battery-business/src/main/java/com/battery/business/mapper/TTestMapper.java
index 6414b6b..327206f 100644
--- a/battery-business/src/main/java/com/battery/business/mapper/TTestMapper.java
+++ b/battery-business/src/main/java/com/battery/business/mapper/TTestMapper.java
@@ -89,6 +89,9 @@
int insertRealtimeRecord(TTestRealtimeRecord record);
String selectRealtimeRecordContentHash(@Param("testId") Long testId, @Param("testBatteryId") Long testBatteryId,
@Param("captureSlotUtc") Date captureSlotUtc);
+ /** 按来源行身份查询已保存的实时记录摘要；逐行接收据此做幂等与冲突判定。 */
+ String selectRealtimeRecordHashBySourceRecord(@Param("testBatteryId") Long testBatteryId,
+ @Param("sourceIdentity") String sourceIdentity, @Param("sourceRecordId") String sourceRecordId);
int insertStepRecord(TTestStepRecord record);
int insertCycleRecord(TTestCycleRecord record);
String selectStepRecordContentHash(@Param("testBatteryId") Long testBatteryId,
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterRuntimeCoordinator.java b/battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterRuntimeCoordinator.java
index d113dff..60d1746 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterRuntimeCoordinator.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterRuntimeCoordinator.java
@@ -2,14 +2,15 @@
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
-import org.springframework.context.annotation.Profile;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.battery.business.service.IDeviceAdapterService;
-/** 在单采集进程内恢复启用实例，并按各自周期执行只读发现。 */
+/**
+ * 在单采集进程内恢复启用实例，并按各自周期执行只读发现。
+ * 不限定 profile：真实来源的发现与接收要在非开发 profile 下运行。
+ */
@Component
-@Profile({ "dev-simulator", "bts" })
public class DeviceAdapterRuntimeCoordinator
{
private final IDeviceAdapterService service;
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterServiceImpl.java
index b42c954..175558e 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterServiceImpl.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterServiceImpl.java
@@ -32,6 +32,10 @@
private static final String DISABLED = "1";
private static final String READ_ONLY_CAPABILITIES =
"{\"discovery\":true,\"realtimeObservation\":true,\"historyDownload\":true,\"control\":false}";
+ /** 适配器类型按存储形态划分；BTS 保留给开发模拟器与将来的协议路线（ADR-0065）。 */
+ private static final List<String> SUPPORTED_TYPES = List.of("BTS", "NEWARE_MYSQL");
+ /** 直连厂商存储的来源，其通道登记清单是长期事实，候选不按时间过期。 */
+ private static final List<String> PERSISTENT_CANDIDATE_TYPES = List.of("NEWARE_MYSQL");
private final TDeviceAdapterMapper mapper;
private final DeviceAdapterRegistry registry;
@@ -136,7 +140,7 @@
registry.synchronize(toRuntimeConfig(adapter));
AdapterDiscoveryResult result = persistence.discoverForReception(registry, new DiscoverySource(
adapter.getAdapterCode(), adapter.getAdapterId(), adapter.getAdapterName(),
- adapter.getPollIntervalMs()));
+ adapter.getPollIntervalMs(), PERSISTENT_CANDIDATE_TYPES.contains(adapter.getAdapterType())));
mapper.updateDiscoverySuccess(adapterId, result.observedAtUtc());
return result;
}
@@ -214,8 +218,9 @@
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 (!"BTS".equals(adapter.getAdapterType()))
+ if (!SUPPORTED_TYPES.contains(adapter.getAdapterType()))
throw error("不支持的适配器类型", DeviceAdapterErrorCodes.UNSUPPORTED_TYPE);
if (!Boolean.TRUE.equals(adapter.getReadOnly()))
throw error("当前里程碑只允许只读设备适配器", DeviceAdapterErrorCodes.INVALID_CONFIG);
@@ -243,7 +248,21 @@
{
return new AdapterInstanceConfig(adapter.getAdapterCode(), adapter.getAdapterType(), adapter.getHost(),
adapter.getPort(), adapter.getPollIntervalMs(), adapter.getTimezoneId(),
- ENABLED.equals(adapter.getStatus()), parseCapabilities(adapter.getCapabilities()));
+ 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)
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceService.java b/battery-business/src/main/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceService.java
index 2db8d34..1fe72af 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceService.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceService.java
@@ -28,6 +28,11 @@
public class DeviceDiscoveryPersistenceService
{
private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30);
+ /**
+ * 直连厂商存储的来源，其通道登记清单是长期存在的事实，没有"过期"的对应语义：
+ * 候选是否出现在界面上只由"已登记且未删除"决定，这个时间只是让候选一直有效。
+ */
+ static final Instant CANDIDATE_NEVER_EXPIRES = Instant.parse("9999-12-31T23:59:59Z");
private final DeviceDiscoveryCandidateStore candidateStore;
private final TDetectionCabinetMapper detectionCabinetMapper;
private final TTestMapper testMapper;
@@ -75,7 +80,8 @@
.map(item -> new CabinetCandidate(item.adapterCode(), item.adapterId(), item.deviceCode(),
item.devtype(), item.devid(), item.ip(), item.mac(), item.name(), item.manufacturer(), item.model(),
item.channels().stream().map(channel -> new ChannelCandidate(channel.identity(), false,
- DeviceWorkStatus.UNOBSERVABLE, failedAt)).toList(), failedAt, item.expiresAt()))
+ DeviceWorkStatus.UNOBSERVABLE, failedAt, null, channel.lastDataAt())).toList(),
+ failedAt, item.expiresAt(), item.lastDataAt()))
.toList());
detectionCabinetMapper.markRegisteredFactsUnobservable(adapterCode);
detectionCabinetMapper.markRegisteredCabinetsOffline(adapterCode);
@@ -108,9 +114,10 @@
LinkedHashMap::new, java.util.stream.Collectors.toList()));
Duration lifetime = Duration.ofMillis(Math.max(MINIMUM_CANDIDATE_LIFETIME.toMillis(),
(long) source.pollIntervalMs() * 3));
+ Instant expiresAt = source.persistentCandidates() ? CANDIDATE_NEVER_EXPIRES : observedAt.plus(lifetime);
List<CabinetCandidate> candidates = grouped.entrySet().stream()
.map(entry -> toCandidate(source, entry.getKey(), entry.getValue(), snapshots,
- observedAt, observedAt.plus(lifetime)))
+ observedAt, expiresAt))
.toList();
candidateStore.recordCabinets(candidates);
if (detected.isEmpty()) markUnavailable(source.adapterCode());
@@ -185,16 +192,27 @@
RealtimeSnapshot snapshot = snapshots.get(item.identity());
DeviceWorkStatus status = snapshot == null || snapshot.workStatus() == null
? DeviceWorkStatus.UNKNOWN : snapshot.workStatus();
- return new ChannelCandidate(item.identity(), item.available(), status, observedAt, snapshots.get(item.identity()));
+ return new ChannelCandidate(item.identity(), item.available(), status, observedAt,
+ snapshots.get(item.identity()), item.lastDataAt());
}).toList();
+ Instant lastDataAt = channels.stream().map(DetectedChannel::lastDataAt)
+ .filter(java.util.Objects::nonNull).max(Instant::compareTo).orElse(null);
return new CabinetCandidate(source.adapterCode(), source.adapterId(),
source.adapterCode() + "-" + key.devtype() + "-" + key.devid(),
key.devtype(), key.devid(), ip, metadata.macAddress(), name,
- metadata.manufacturer(), metadata.model(), channelCandidates, observedAt, expiresAt);
+ metadata.manufacturer(), metadata.model(), channelCandidates, observedAt, expiresAt, lastDataAt);
}
/** 发现来源只携带候选标识和过期计算所需配置，不复用数据库实体。 */
- public record DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs) { }
+ public record DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs,
+ boolean persistentCandidates)
+ {
+ /** 兼容按轮询过期计算候选的既有来源。 */
+ public DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs)
+ {
+ this(adapterCode, adapterId, adapterName, pollIntervalMs, false);
+ }
+ }
private record DiscoveryObservation(List<DetectedChannel> channels,
Map<ChannelIdentity, RealtimeSnapshot> snapshots) { }
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java b/battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java
new file mode 100644
index 0000000..067c86b
--- /dev/null
+++ b/battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java
@@ -0,0 +1,130 @@
+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.DeviceAdapterRegistry;
+
+/**
+ * 把取数请求按通道身份里的适配器编码送到它所属的来源实例。
+ *
+ * <p>这是持久化与实时展示路径上唯一的取数入口：同一个进程里可以注册多个来源实例，
+ * 每个通道的请求只送到它自己的实例；实例缺失或未提供测试数据观测契约时明确失败，
+ * 不回退到其他实例。发现不走这里，它按实例在 {@link DeviceDiscoveryPersistenceService} 中执行。
+ *
+ * <p>声明为 {@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<DetectedChannel> discoverChannels()
+ {
+ throw new UnsupportedOperationException("设备发现按来源实例执行，不在聚合适配器上执行");
+ }
+
+ /** 一次观测里的通道可能来自多个来源实例，逐个按各自的编码请求，不能按第一个通道决定全部。 */
+ @Override
+ public Map<ChannelIdentity, DeviceObservation> observe(Set<ChannelIdentity> channels)
+ {
+ if (channels == null || channels.isEmpty()) return Map.of();
+ Map<ChannelIdentity, DeviceObservation> result = new java.util.LinkedHashMap<>();
+ for (Map.Entry<String, List<ChannelIdentity>> group : groupBySource(channels).entrySet())
+ {
+ Set<ChannelIdentity> groupChannels = new java.util.LinkedHashSet<>(group.getValue());
+ result.putAll(registry.executeTestData(group.getKey(), adapter -> adapter.observe(groupChannels)));
+ }
+ return Map.copyOf(result);
+ }
+
+ private static Map<String, List<ChannelIdentity>> groupBySource(Set<ChannelIdentity> channels)
+ {
+ Map<String, List<ChannelIdentity>> grouped = new java.util.LinkedHashMap<>();
+ for (ChannelIdentity channel : channels)
+ grouped.computeIfAbsent(channel.adapterCode(), ignored -> new java.util.ArrayList<>()).add(channel);
+ return grouped;
+ }
+
+ @Override
+ public CabinetRealtime latestCabinet(ChannelIdentity channel)
+ {
+ return registry.executeTestData(channel.adapterCode(), adapter -> adapter.latestCabinet(channel));
+ }
+
+ @Override
+ public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
+ {
+ // 采集器测量属于某个来源实例的辅助测量，按该实例请求；实例缺失时明确失败。
+ // 来源编码为空说明测试电池连检测柜绑定都没有，此时来源未知：不猜也不回退，报无法请求。
+ if (adapterCode == null || adapterCode.isBlank()) return null;
+ return registry.executeTestData(adapterCode, adapter -> adapter.latestCollector(adapterCode, collectorCode));
+ }
+
+ @Override
+ public List<StepRecord> stepRecords(ChannelIdentity channel)
+ {
+ return registry.executeTestData(channel.adapterCode(), adapter -> adapter.stepRecords(channel));
+ }
+
+ @Override
+ public List<CycleRecord> cycleRecords(ChannelIdentity channel)
+ {
+ return registry.executeTestData(channel.adapterCode(), adapter -> adapter.cycleRecords(channel));
+ }
+
+ @Override
+ public long persistenceIntervalMillis(ChannelIdentity channel)
+ {
+ return registry.executeTestData(channel.adapterCode(), adapter -> adapter.persistenceIntervalMillis(channel));
+ }
+
+ @Override
+ public TestPlan testPlan(ChannelIdentity channel)
+ {
+ return registry.executeTestData(channel.adapterCode(), adapter -> adapter.testPlan(channel));
+ }
+
+ @Override
+ public SampleWatermark currentWatermark(ChannelIdentity channel)
+ {
+ return registry.executeTestData(channel.adapterCode(), adapter -> adapter.currentWatermark(channel));
+ }
+
+ @Override
+ public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
+ {
+ return registry.executeTestData(channel.adapterCode(), adapter -> adapter.samplesSince(channel, watermark));
+ }
+}
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/TDetectionCabinetServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/TDetectionCabinetServiceImpl.java
index de81e9c..9c485e1 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/TDetectionCabinetServiceImpl.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/TDetectionCabinetServiceImpl.java
@@ -18,6 +18,7 @@
import com.battery.business.domain.TDetectionChannel;
import com.battery.business.domain.vo.DiscoveredCabinetVo;
import com.battery.business.mapper.TDetectionCabinetMapper;
+import com.battery.business.service.IDeviceAdapterService;
import com.battery.business.service.ITDetectionCabinetService;
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore;
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
@@ -39,6 +40,7 @@
private final TDetectionCabinetMapper detectionCabinetMapper;
private final DeviceDiscoveryCandidateStore candidateStore;
+ private final IDeviceAdapterService adapterService;
/**
* 构造检测柜业务服务。
@@ -48,8 +50,42 @@
public TDetectionCabinetServiceImpl(TDetectionCabinetMapper detectionCabinetMapper,
DeviceDiscoveryCandidateStore candidateStore)
{
+ this(detectionCabinetMapper, candidateStore, null);
+ }
+
+ /**
+ * 构造检测柜业务服务。
+ *
+ * @param detectionCabinetMapper 检测柜数据访问对象
+ * @param candidateStore 发现候选存储
+ * @param adapterService 适配器实例服务；为空时列表不展示数据来源名称
+ */
+ @org.springframework.beans.factory.annotation.Autowired
+ public TDetectionCabinetServiceImpl(TDetectionCabinetMapper detectionCabinetMapper,
+ DeviceDiscoveryCandidateStore candidateStore,
+ @org.springframework.lang.Nullable IDeviceAdapterService adapterService)
+ {
this.detectionCabinetMapper = detectionCabinetMapper;
this.candidateStore = candidateStore;
+ this.adapterService = adapterService;
+ }
+
+ /** 数据来源名称按适配器编码解析；代码装配实例没有数据库记录时退回展示编码。 */
+ private Map<String, String> adapterNames()
+ {
+ if (adapterService == null) return Map.of();
+ return adapterService.selectList(new com.battery.business.domain.TDeviceAdapter()).stream()
+ .filter(item -> item.getAdapterCode() != null && item.getAdapterName() != null)
+ .collect(java.util.stream.Collectors.toMap(
+ com.battery.business.domain.TDeviceAdapter::getAdapterCode,
+ com.battery.business.domain.TDeviceAdapter::getAdapterName, (left, right) -> right));
+ }
+
+ /** 适配器编码可能缺失（历史登记），缺失时不查询来源名称。 */
+ private void applyAdapterName(TDetectionCabinet cabinet, Map<String, String> adapterNames)
+ {
+ if (cabinet == null || cabinet.getAdapterCode() == null) return;
+ cabinet.setAdapterName(adapterNames.get(cabinet.getAdapterCode()));
}
/**
@@ -66,6 +102,7 @@
{
throw new ServiceException("检测柜不存在或已删除");
}
+ applyAdapterName(detectionCabinet, adapterNames());
return applyRealtimeState(detectionCabinet, candidatesByCode());
}
@@ -87,8 +124,10 @@
detectionCabinet.setOnlineStatus(null);
detectionCabinet.setTestStatus(null);
Map<String, CabinetCandidate> candidates = candidatesByCode();
+ Map<String, String> adapterNames = adapterNames();
return detectionCabinetMapper.selectDetectionCabinetList(detectionCabinet).stream()
.map(item -> applyRealtimeState(item, candidates))
+ .peek(item -> applyAdapterName(item, adapterNames))
.filter(item -> StringUtils.isBlank(requestedOnlineStatus)
|| requestedOnlineStatus.equals(item.getOnlineStatus()))
.filter(item -> StringUtils.isBlank(requestedTestStatus)
@@ -97,8 +136,10 @@
}
/**
- * 读取未过期且未登记的候选，离线设备仍保留身份并可选择。
- * 发现只返回设备事实，不从通道工作状态推断系统测试关系。
+ * 读取未登记且未删除的候选，离线设备仍保留身份并可选择。
+ * 候选不按时间过期（厂商存储里的通道登记是长期事实），只按"已登记且未删除"排除；
+ * 每条附带最近数据时间并按它倒序，便于操作员判断哪几台真的在产出数据。
+ * 发现只返回设备事实，不从通道工作状态推断系统测试关系，也不自动判定"在用"。
*
* @return 可供选择的检测柜设备摘要
*/
@@ -110,7 +151,11 @@
.filter(candidate -> !registeredCodes.contains(candidate.deviceCode()))
.map(candidate -> new DiscoveredCabinetVo(
candidate.deviceCode(), candidate.ip(), candidate.mac(), candidate.name(),
- candidate.manufacturer(), candidate.model(), candidate.channels().size(), candidate.online() ? "在线" : "离线"))
+ candidate.manufacturer(), candidate.model(), candidate.channels().size(),
+ candidate.online() ? "在线" : "离线", candidate.lastDataAt()))
+ .sorted(java.util.Comparator.comparing(DiscoveredCabinetVo::lastDataAt,
+ java.util.Comparator.nullsLast(java.util.Comparator.reverseOrder()))
+ .thenComparing(DiscoveredCabinetVo::deviceCode))
.toList();
}
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java
index 0ee9d4d..f9a7270 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java
@@ -191,6 +191,10 @@
private Map<String, Object> snapshot(ChannelCandidate channel)
{
Map<String, Object> value = new LinkedHashMap<>();
+ // 通道工作状态是独立设备事实：读库来源下不可观测，界面把它当正常取值展示而不是异常。
+ value.put("workStatus", channel.workStatus() == null ? null : channel.workStatus().name());
+ // 最近数据时间取来源里最近一次采样行的设备时间，与展示轮询节奏无关。
+ value.put("lastDataAt", channel.lastDataAt() == null ? null : channel.lastDataAt().toEpochMilli());
if (channel.latestSnapshot() == null) return value;
value.put("source", "检测柜");
value.put("observedAt", channel.latestSnapshot().receivedAt().toEpochMilli());
@@ -232,7 +236,8 @@
return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
online ? "ONLINE" : "OFFLINE", Map.of());
}
- var collector = adapter.latestCollector(collectorCode);
+ var collector = adapter.latestCollector(
+ battery.getBinding() == null ? null : battery.getBinding().getAdapterCode(), collectorCode);
if (collector != null)
{
Map<String, Object> value = new LinkedHashMap<>();
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
index 9ef74b0..0a2f7f3 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
@@ -23,6 +23,7 @@
import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
import com.battery.device.api.TestDataObservationAdapter.CollectorMeasurement;
import com.battery.device.api.TestDataObservationAdapter.CollectorRealtime;
+import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.business.domain.TTest;
@@ -47,6 +48,9 @@
private final Clock clock;
private final Map<String, Instant> lastSnapshotAt = new java.util.concurrent.ConcurrentHashMap<>();
private final Map<Long, Instant> lastRealtimeCheckAt = new java.util.concurrent.ConcurrentHashMap<>();
+ /** 每个测试电池各自的接收水位；只存在内存中，重启后按来源里的最新一行重新取水位。 */
+ private final Map<String, SampleCursor> sampleCursors = new java.util.concurrent.ConcurrentHashMap<>();
+
@Autowired
public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable TestDataObservationAdapter adapter,
@@ -95,11 +99,127 @@
private void saveRealtime(TTest test, TTestBattery battery)
{
- // 柜端和采集器端分别判断成功；同一槽位只写一行，失败来源的身份、时间和指标保持为空。
if (adapter == null) return;
TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
+ // 来源以设备数据行为接收单位时，柜端事实逐行落库（同秒多行都保留），槽位只留给采集器端。
+ boolean rowBased = persistCabinetSamples(test, battery, binding);
+ saveRealtimeSlot(test, battery, binding, !rowBased);
+ }
+
+ /**
+ * 按来源水位逐行接收柜端事实。水位是"开始接收"或"继续接收"这一刻来源里的最新一行，
+ * 因此水位之前的数据行即使仍在来源里也不进本测试，暂停期间的空档也不会被补回来。
+ *
+ * @return 该来源是否按设备数据行接收；来源不支持或被拒绝时返回 false，调用方回落到轮询槽位
+ */
+ private boolean persistCabinetSamples(TTest test, TTestBattery battery, TTestChannelBinding binding)
+ {
+ String key = test.getTestId() + ":" + battery.getTestBatteryId();
+ Instant boundary = test.getReceptionBoundaryAtUtc() == null ? null : test.getReceptionBoundaryAtUtc().toInstant();
+ SampleCursor cursor = sampleCursors.computeIfAbsent(key, ignored -> new SampleCursor());
+ synchronized (cursor)
+ {
+ // 开始接收与继续接收都会推进接收边界：边界变化即重新取水位，暂停期间的数据不补采。
+ if (!java.util.Objects.equals(cursor.boundary, boundary))
+ {
+ cursor.boundary = boundary; cursor.watermark = null; cursor.rowBased = null; cursor.sampledAt = null;
+ }
+ if (Boolean.FALSE.equals(cursor.rowBased)) return false;
+ ChannelIdentity channel = channelIdentity(binding);
+ Instant now = clock.instant();
+ long interval = persistenceInterval(binding);
+ // 厂商写库节奏是 60 秒，取数不必更快；展示节奏由控制台轮询单独承担。
+ if (cursor.sampledAt != null && now.isBefore(cursor.sampledAt.plusMillis(interval)))
+ return Boolean.TRUE.equals(cursor.rowBased);
+ try
+ {
+ if (cursor.watermark == null)
+ {
+ cursor.sampledAt = now;
+ SampleWatermark watermark = adapter.currentWatermark(channel);
+ if (watermark == null)
+ {
+ // 来源不按设备数据行接收（例如开发模拟器）：柜端事实继续走轮询槽位。
+ cursor.rowBased = false;
+ return false;
+ }
+ cursor.watermark = watermark;
+ cursor.rowBased = true;
+ return true;
+ }
+ List<CabinetRealtime> rows = adapter.samplesSince(channel, cursor.watermark);
+ cursor.sampledAt = now;
+ for (CabinetRealtime row : rows)
+ {
+ insertCabinetSample(test, battery, binding, row);
+ cursor.watermark = new SampleWatermark(row.deviceTime(), row.sequenceId());
+ }
+ return true;
+ }
+ catch (RuntimeException exception)
+ {
+ // 实例缺失、来源不可达或读取失败：明确失败并留痕，下一轮重试；不回退到其他实例。
+ log.warn("柜端逐行接收失败 testId={}, bindingId={}, adapterCode={}", test.getTestId(),
+ binding.getBindingId(), binding.getAdapterCode(), exception);
+ cursor.sampledAt = now;
+ return Boolean.TRUE.equals(cursor.rowBased);
+ }
+ }
+ }
+
+ /**
+ * 把一行设备事实写成一条实时记录；来源行身份保证重复接收幂等、同身份不同内容报冲突。
+ * 归属边界由水位在设备时间里判定（见 {@link #persistCabinetSamples}），
+ * 这里不用服务器取数时间反推设备时间。
+ */
+ private void insertCabinetSample(TTest test, TTestBattery battery, TTestChannelBinding binding, CabinetRealtime row)
+ {
+ if (row.deviceTime() == null || row.serverReceivedAt() == null) return;
+ Date observedAt = Date.from(row.serverReceivedAt());
+ evaluateCabinet(test, binding, "VOLTAGE", BigDecimal.valueOf(row.voltageV()), "V", observedAt);
+ evaluateCabinet(test, binding, "CURRENT", BigDecimal.valueOf(row.currentA()), "A", observedAt);
+ evaluateCabinet(test, binding, "CAPACITY", BigDecimal.valueOf(row.capacityAh()), "Ah", observedAt);
+ evaluateCabinet(test, binding, "ENERGY", BigDecimal.valueOf(row.energyWh()), "Wh", observedAt);
+ String identity = sourceIdentity(binding);
+ String recordId = row.sourceRecordId() == null || row.sourceRecordId().isBlank()
+ ? "cabinet:" + row.sequenceId() : row.sourceRecordId();
+ TTestRealtimeRecord value = new TTestRealtimeRecord();
+ value.setTestId(test.getTestId()); value.setTestBatteryId(battery.getTestBatteryId());
+ value.setBindingId(binding.getBindingId()); value.setAdapterCode(binding.getAdapterCode());
+ // 设备数据行不是系统轮询槽位，槽位留空：同秒多行因此可以各自成立（ADR-0051 的槽位可空机制）。
+ value.setCaptureSlotUtc(null); value.setServerReceivedAtUtc(observedAt);
+ value.setCabinetDevtype(binding.getDevtype()); value.setCabinetDevid(binding.getDevid());
+ value.setCabinetSubdevid(binding.getSubdevid()); value.setCabinetChlid(binding.getChlid());
+ value.setCabinetAdapterTimezoneId(row.timezoneId());
+ value.setCabinetSequenceId(row.sequenceId());
+ value.setCabinetDeviceTimeUtc(Date.from(row.deviceTime()));
+ value.setCabinetDeviceTimeRaw(row.deviceTimeRaw());
+ value.setCycleNo(row.cycleNo()); value.setStepNo(row.stepNo()); value.setStepType(row.stepType());
+ value.setStepName(row.stepName());
+ value.setVoltageV(BigDecimal.valueOf(row.voltageV())); value.setCurrentA(BigDecimal.valueOf(row.currentA()));
+ value.setCapacityAh(BigDecimal.valueOf(row.capacityAh())); value.setEnergyWh(BigDecimal.valueOf(row.energyWh()));
+ Map<String, Object> extension = new LinkedHashMap<>();
+ if (row.relativeTimeSeconds() != null) extension.put("relativeTimeSeconds", row.relativeTimeSeconds());
+ value.setExtensionMeasurementsJson(JSON.toJSONString(extension));
+ value.setSourceIdentity(identity); value.setSourceRecordId(recordId);
+ value.setContentSha256(hash(value));
+ // 重复接收的同内容记录幂等忽略；同身份但内容不同时保留原事实，不覆盖。
+ String existing = mapper.selectRealtimeRecordHashBySourceRecord(battery.getTestBatteryId(), identity, recordId);
+ if (existing != null)
+ {
+ if (!existing.equalsIgnoreCase(HexFormat.of().formatHex(value.getContentSha256())))
+ log.warn("实时记录内容冲突，拒绝覆盖 testId={}, testBatteryId={}, sourceRecordId={}",
+ test.getTestId(), battery.getTestBatteryId(), recordId);
+ return;
+ }
+ mapper.insertRealtimeRecord(value);
+ }
+
+ private void saveRealtimeSlot(TTest test, TTestBattery battery, TTestChannelBinding binding, boolean includeCabinet)
+ {
+ // 柜端和采集器端分别判断成功；同一槽位只写一行，失败来源的身份、时间和指标保持为空。
Instant now = clock.instant(); long interval = persistenceInterval(binding); String key = test.getTestId() + ":" + battery.getTestBatteryId();
- CabinetRealtime cabinet = latestCabinet(binding);
+ CabinetRealtime cabinet = includeCabinet ? latestCabinet(binding) : null;
if (cabinet != null && !accepted(test, cabinet.deviceTime(), cabinet.serverReceivedAt())) cabinet = null;
CollectorRealtime collector = latestCollector(battery);
Map<String, CollectorMeasurement> mapped = mappedCollectorMetrics(test, battery, collector);
@@ -192,8 +312,9 @@
}
}
- private CabinetRealtime latestCabinet(TTestChannelBinding b) { try { return adapter.latestCabinet(new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid())); } catch (RuntimeException ex) { return null; } }
- private CollectorRealtime latestCollector(TTestBattery b) { if (b.getCollectorId() == null) return null; try { return adapter.latestCollector(mapper.selectCollectorCode(b.getCollectorId())); } catch (RuntimeException ex) { return null; } }
+ private CabinetRealtime latestCabinet(TTestChannelBinding b) { try { return adapter.latestCabinet(channelIdentity(b)); } catch (RuntimeException ex) { return null; } }
+ private static ChannelIdentity channelIdentity(TTestChannelBinding b) { return new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid()); }
+ private CollectorRealtime latestCollector(TTestBattery b) { if (b.getCollectorId() == null || b.getBinding() == null) return null; try { return adapter.latestCollector(b.getBinding().getAdapterCode(), mapper.selectCollectorCode(b.getCollectorId())); } catch (RuntimeException ex) { return null; } }
private Map<String, CollectorMeasurement> mappedCollectorMetrics(TTest test, TTestBattery battery, CollectorRealtime source)
{
Map<String, CollectorMeasurement> result = new LinkedHashMap<>();
@@ -222,7 +343,7 @@
.ifPresent(sequence -> parts.add("collector:" + sequence));
return String.join(":", parts);
}
- private long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid()))); } catch (RuntimeException ex) { return 60000L; } }
+ private long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(channelIdentity(b))); } catch (RuntimeException ex) { return 60000L; } }
private static String sourceIdentity(TTestChannelBinding b) { return String.join(":", String.valueOf(b.getAdapterCode()), String.valueOf(b.getDevtype()), String.valueOf(b.getDevid()), String.valueOf(b.getSubdevid()), String.valueOf(b.getChlid())); }
private static String timezone(TTestChannelBinding b) { return b.getAdapterTimezoneId() == null || b.getAdapterTimezoneId().isBlank() ? "Asia/Hong_Kong" : b.getAdapterTimezoneId(); }
private static String blankToNull(String value) { return value == null || value.isBlank() ? null : value; }
@@ -241,4 +362,13 @@
try { return MessageDigest.getInstance("SHA-256").digest(JSON.toJSONBytes(content)); }
catch (Exception ex) { throw new IllegalStateException("无法计算测试数据内容摘要", ex); }
}
+
+ /** 单个测试电池的逐行接收状态：接收边界、来源水位、是否按行接收与上次取数时间。 */
+ private static final class SampleCursor
+ {
+ private Instant boundary;
+ private SampleWatermark watermark;
+ private Boolean rowBased;
+ private Instant sampledAt;
+ }
}
diff --git a/battery-business/src/main/resources/mapper/battery/TDeviceAdapterMapper.xml b/battery-business/src/main/resources/mapper/battery/TDeviceAdapterMapper.xml
index 5b75ff5..e041146 100644
--- a/battery-business/src/main/resources/mapper/battery/TDeviceAdapterMapper.xml
+++ b/battery-business/src/main/resources/mapper/battery/TDeviceAdapterMapper.xml
@@ -11,6 +11,7 @@
<result property="pollIntervalMs" column="poll_interval_ms"/>
<result property="timezoneId" column="timezone_id"/>
<result property="capabilities" column="capabilities"/>
+ <result property="sourceConfig" column="source_config"/>
<result property="readOnly" column="read_only"/>
<result property="status" column="status"/>
<result property="connectionStatus" column="connection_status"/>
@@ -28,7 +29,7 @@
<sql id="adapterColumns">
adapter_id, adapter_code, adapter_name, adapter_type, host, port, poll_interval_ms, timezone_id,
- capabilities, (json_extract(capabilities, '$.control') = false) read_only, status,
+ capabilities, source_config, (json_extract(capabilities, '$.control') = false) read_only, status,
connection_status, last_error_code, last_error_message, last_attempt_at_utc, last_observed_at_utc,
create_by, create_time, update_by, update_time, remark, del_flag
</sql>
@@ -61,27 +62,29 @@
<!-- 应用装配层提供托管配置；持久化层仅负责按稳定编码幂等恢复。 -->
<insert id="upsertManagedDeviceAdapter" parameterType="TDeviceAdapter">
insert into t_device_adapter(adapter_code, adapter_name, adapter_type, host, port, poll_interval_ms,
- timezone_id, capabilities, status, connection_status, create_by, create_time, update_by, update_time,
+ timezone_id, capabilities, source_config, status, connection_status, create_by, create_time, update_by, update_time,
remark, del_flag)
values(#{adapterCode}, #{adapterName}, #{adapterType}, #{host}, #{port}, #{pollIntervalMs}, #{timezoneId},
- #{capabilities}, '0', 'OFFLINE', 'system', sysdate(), 'system', sysdate(), #{remark}, 0)
+ #{capabilities}, cast(#{sourceConfig} as json), '0', 'OFFLINE', 'system', sysdate(), 'system', sysdate(), #{remark}, 0)
on duplicate key update adapter_name=values(adapter_name), adapter_type=values(adapter_type),
host=values(host), port=values(port), poll_interval_ms=values(poll_interval_ms),
- timezone_id=values(timezone_id), capabilities=values(capabilities), status='0',
+ timezone_id=values(timezone_id), capabilities=values(capabilities), source_config=values(source_config),
+ status='0',
connection_status='OFFLINE', last_error_code=null, last_error_message=null,
update_by='system', update_time=sysdate(), remark=values(remark), del_flag=0
</insert>
<insert id="insertDeviceAdapter" parameterType="TDeviceAdapter" useGeneratedKeys="true" keyProperty="adapterId">
insert into t_device_adapter(adapter_code, adapter_name, adapter_type, host, port, poll_interval_ms,
- timezone_id, capabilities, status, connection_status, create_by, create_time, remark, del_flag)
+ timezone_id, capabilities, source_config, status, connection_status, create_by, create_time, remark, del_flag)
values(#{adapterCode}, #{adapterName}, #{adapterType}, #{host}, #{port}, #{pollIntervalMs},
- #{timezoneId}, #{capabilities}, #{status}, 'OFFLINE', #{createBy}, sysdate(), #{remark}, 0)
+ #{timezoneId}, #{capabilities}, cast(#{sourceConfig} as json), #{status}, 'OFFLINE', #{createBy}, sysdate(), #{remark}, 0)
</insert>
<update id="updateDeviceAdapter" parameterType="TDeviceAdapter">
update t_device_adapter set adapter_name=#{adapterName}, adapter_type=#{adapterType}, host=#{host}, port=#{port},
poll_interval_ms=#{pollIntervalMs}, timezone_id=#{timezoneId}, capabilities=#{capabilities},
+ source_config=cast(#{sourceConfig} as json),
connection_status='OFFLINE', last_error_code=null, last_error_message=null,
update_by=#{updateBy}, update_time=sysdate()
where adapter_id=#{adapterId} and del_flag=0
diff --git a/battery-business/src/main/resources/mapper/battery/TTestMapper.xml b/battery-business/src/main/resources/mapper/battery/TTestMapper.xml
index 00e2240..79bf3c5 100644
--- a/battery-business/src/main/resources/mapper/battery/TTestMapper.xml
+++ b/battery-business/src/main/resources/mapper/battery/TTestMapper.xml
@@ -382,6 +382,12 @@
where test_id = #{testId} and test_battery_id = #{testBatteryId} and capture_slot_utc = #{captureSlotUtc}
limit 1
</select>
+ <select id="selectRealtimeRecordHashBySourceRecord" resultType="string">
+ select lower(hex(content_sha256)) from t_test_realtime_record
+ where test_battery_id = #{testBatteryId} and source_identity = #{sourceIdentity}
+ and source_record_id = #{sourceRecordId}
+ limit 1
+ </select>
<insert id="insertStepRecord" parameterType="com.battery.business.domain.TTestStepRecord" useGeneratedKeys="true" keyProperty="stepRecordId">
insert ignore into t_test_step_record
(test_id, test_battery_id, binding_id, adapter_code, source_identity, source_record_id, dedupe_key, adapter_timezone_id,
diff --git a/battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java b/battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java
index 3dd9033..ef44f5d 100644
--- a/battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java
+++ b/battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java
@@ -12,8 +12,12 @@
/** 返回指定检测柜通道最近一次实时观测；没有成功观测时返回空值。 */
CabinetRealtime latestCabinet(ChannelIdentity channel);
- /** 返回采集器独立实时快照；不得通过历史下载补采。 */
- CollectorRealtime latestCollector(String collectorCode);
+ /**
+ * 返回采集器独立实时快照；不得通过历史下载补采。
+ * 采集器测量按"同一来源的辅助测量"表达（ADR-0064），所以必须指名是哪个来源实例，
+ * 否则多实例下无法确定该向哪台机器请求。
+ */
+ CollectorRealtime latestCollector(String adapterCode, String collectorCode);
/** 返回设备直接提供的工步统计，不从实时采样推导。 */
List<StepRecord> stepRecords(ChannelIdentity channel);
@@ -27,14 +31,49 @@
/** 返回设备计划方案；设备未提供时保持为空，不推导计划。 */
default TestPlan testPlan(ChannelIdentity channel) { return null; }
+ /**
+ * 读取通道当前的接收水位。水位是来源里已存在的最新一行（设备绝对时间 + 协议序号），
+ * 用于把"开始接收"和"继续接收"切在这一时刻之后：水位之前的数据行即使仍在来源里也不进本测试。
+ *
+ * <p>返回空值表示该来源不按设备数据行接收（例如由系统轮询产生槽位的来源），此时不启用逐行接收。
+ * 设备时间为空的水位表示这一刻来源里没有任何可用行，之后出现的行全部属于本次接收。
+ * 来源不可达时必须抛出异常，不能返回空值——空值会被误当作"来源不按行接收"。
+ */
+ default SampleWatermark currentWatermark(ChannelIdentity channel) { return null; }
+
+ /**
+ * 返回严格晚于水位的设备数据行，按设备时间与协议序号升序。
+ * 同一绝对秒内的多行各自成立，调用方不得按秒合并。
+ * 水位为空表示调用方尚未取得水位，此时返回空列表：不把来源里的设备历史并入本次接收。
+ */
+ default List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark) { return List.of(); }
+
+ /** 接收水位：设备绝对时间与协议序号的二元组；序号在同一厂商测试内有效。 */
+ record SampleWatermark(Instant deviceTime, long sequenceId) { }
+
record TestPlan(int currentCycleNo, int currentStepNo, int plannedCycleCount, List<PlanRow> rows) { }
record PlanRow(String rowType, int stepNo, Integer cyclePosition, String stepName, double durationSeconds,
Double voltageV, String executionStatus, Integer cycleCount, Long recordCount) { }
- /** 柜端成功观测；相对时长使用设备提供的秒数，缺失不按采样时间推算。 */
+ /**
+ * 柜端成功观测；相对时长使用设备提供的秒数，缺失不按采样时间推算。
+ * {@code sourceRecordId} 是来源行的稳定身份（如厂商测试编号加协议序号），
+ * 供幂等与冲突判定使用；来源不提供时保持为空，调用方退化为按协议序号识别。
+ */
record CabinetRealtime(long sequenceId, Instant deviceTime, double currentA, double voltageV,
double capacityAh, double energyWh, int cycleNo, int stepNo, String stepName, String stepType,
- Instant serverReceivedAt, String deviceTimeRaw, String timezoneId, Long relativeTimeSeconds) { }
+ Instant serverReceivedAt, String deviceTimeRaw, String timezoneId, Long relativeTimeSeconds,
+ String sourceRecordId)
+ {
+ /** 兼容不携带来源行身份的既有适配器。 */
+ public CabinetRealtime(long sequenceId, Instant deviceTime, double currentA, double voltageV,
+ double capacityAh, double energyWh, int cycleNo, int stepNo, String stepName, String stepType,
+ Instant serverReceivedAt, String deviceTimeRaw, String timezoneId, Long relativeTimeSeconds)
+ {
+ this(sequenceId, deviceTime, currentA, voltageV, capacityAh, energyWh, cycleNo, stepNo, stepName,
+ stepType, serverReceivedAt, deviceTimeRaw, timezoneId, relativeTimeSeconds, null);
+ }
+ }
record CollectorRealtime(String collectorCode, boolean online, int sampleIntervalMs,
List<CollectorMeasurement> measurements, String timezoneId)
diff --git a/battery-device-api/src/main/java/com/battery/device/api/discovery/DeviceDiscoveryCandidateStore.java b/battery-device-api/src/main/java/com/battery/device/api/discovery/DeviceDiscoveryCandidateStore.java
index 54e5663..733965b 100644
--- a/battery-device-api/src/main/java/com/battery/device/api/discovery/DeviceDiscoveryCandidateStore.java
+++ b/battery-device-api/src/main/java/com/battery/device/api/discovery/DeviceDiscoveryCandidateStore.java
@@ -43,22 +43,43 @@
}
}
+ /**
+ * 未登记的候选设备。{@code lastDataAt} 取该设备各通道最近的采样行时间，
+ * 供操作员判断哪几台设备现在真的在产出数据；它不改变候选的取舍，登记范围由操作员决定。
+ */
record CabinetCandidate(String adapterCode, Long adapterId, String deviceCode, int devtype, int devid,
String ip, String mac, String name, String manufacturer, String model,
- List<ChannelCandidate> channels, Instant observedAt, Instant expiresAt)
+ List<ChannelCandidate> channels, Instant observedAt, Instant expiresAt, Instant lastDataAt)
{
public CabinetCandidate { channels = List.copyOf(channels); }
public boolean online() { return channels.stream().anyMatch(ChannelCandidate::online); }
+
+ /** 兼容不含最近数据时间的既有候选。 */
+ public CabinetCandidate(String adapterCode, Long adapterId, String deviceCode, int devtype, int devid,
+ String ip, String mac, String name, String manufacturer, String model,
+ List<ChannelCandidate> channels, Instant observedAt, Instant expiresAt)
+ {
+ this(adapterCode, adapterId, deviceCode, devtype, devid, ip, mac, name, manufacturer, model,
+ channels, observedAt, expiresAt, null);
+ }
}
+ /** 通道候选；{@code lastDataAt} 是该通道来源里最近一次采样行的设备时间。 */
record ChannelCandidate(ChannelIdentity identity, boolean online, DeviceWorkStatus workStatus,
- Instant observedAt, RealtimeSnapshot latestSnapshot)
+ Instant observedAt, RealtimeSnapshot latestSnapshot, Instant lastDataAt)
{
/** 兼容仅有状态的旧发现候选。 */
public ChannelCandidate(ChannelIdentity identity, boolean online, DeviceWorkStatus workStatus,
Instant observedAt)
{
- this(identity, online, workStatus, observedAt, null);
+ this(identity, online, workStatus, observedAt, null, null);
+ }
+
+ /** 兼容不含最近数据时间的既有候选。 */
+ public ChannelCandidate(ChannelIdentity identity, boolean online, DeviceWorkStatus workStatus,
+ Instant observedAt, RealtimeSnapshot latestSnapshot)
+ {
+ this(identity, online, workStatus, observedAt, latestSnapshot, null);
}
}
diff --git a/battery-device-api/src/main/java/com/battery/device/api/model/DetectedChannel.java b/battery-device-api/src/main/java/com/battery/device/api/model/DetectedChannel.java
index f244edb..d955ad3 100644
--- a/battery-device-api/src/main/java/com/battery/device/api/model/DetectedChannel.java
+++ b/battery-device-api/src/main/java/com/battery/device/api/model/DetectedChannel.java
@@ -1,10 +1,23 @@
package com.battery.device.api.model;
+import java.time.Instant;
+
+/**
+ * 发现结果中的一个通道。
+ * {@code lastDataAt} 是来源里最近一次采样行的设备时间；它是"最近是否有新采样行写入"这一条依据的取值，
+ * 与设备结束时间或通道水位线列无关。来源不提供该事实时保持为空。
+ */
public record DetectedChannel(ChannelIdentity identity, String serverIp, boolean available,
- DetectedDeviceMetadata deviceMetadata)
+ DetectedDeviceMetadata deviceMetadata, Instant lastDataAt)
{
public DetectedChannel(ChannelIdentity identity, String serverIp, boolean available)
{
- this(identity, serverIp, available, DetectedDeviceMetadata.empty());
+ this(identity, serverIp, available, DetectedDeviceMetadata.empty(), null);
+ }
+
+ public DetectedChannel(ChannelIdentity identity, String serverIp, boolean available,
+ DetectedDeviceMetadata deviceMetadata)
+ {
+ this(identity, serverIp, available, deviceMetadata, null);
}
}
diff --git a/battery-device-api/src/main/java/com/battery/device/api/registry/AdapterInstanceConfig.java b/battery-device-api/src/main/java/com/battery/device/api/registry/AdapterInstanceConfig.java
index 389645a..fc99987 100644
--- a/battery-device-api/src/main/java/com/battery/device/api/registry/AdapterInstanceConfig.java
+++ b/battery-device-api/src/main/java/com/battery/device/api/registry/AdapterInstanceConfig.java
@@ -2,6 +2,7 @@
import java.time.ZoneId;
import java.util.Locale;
+import java.util.Map;
import java.util.Objects;
import com.battery.device.api.model.AdapterCapabilities;
@@ -14,8 +15,12 @@
int pollIntervalMs,
String timezoneId,
boolean enabled,
- AdapterCapabilities capabilities)
+ AdapterCapabilities capabilities,
+ Map<String, String> sourceConfig)
{
+ /**
+ * 来源形态配置按存储形态互不相交，只承载非敏感项；连接口令由环境变量或启动参数提供，不下沉到数据库。
+ */
public AdapterInstanceConfig
{
adapterCode = requireText(adapterCode, "adapterCode");
@@ -23,6 +28,7 @@
host = requireText(host, "host");
timezoneId = requireText(timezoneId, "timezoneId");
capabilities = Objects.requireNonNull(capabilities, "capabilities");
+ sourceConfig = sourceConfig == null ? Map.of() : Map.copyOf(sourceConfig);
if (port < 1 || port > 65535) throw new IllegalArgumentException("port must be between 1 and 65535");
if (pollIntervalMs < 1) throw new IllegalArgumentException("pollIntervalMs must be positive");
ZoneId.of(timezoneId);
@@ -30,6 +36,19 @@
throw new IllegalArgumentException("MVP adapter must support discovery and remain read-only");
}
+ /** 兼容不含来源形态配置的既有装配。 */
+ public AdapterInstanceConfig(String adapterCode, String adapterType, String host, int port, int pollIntervalMs,
+ String timezoneId, boolean enabled, AdapterCapabilities capabilities)
+ {
+ this(adapterCode, adapterType, host, port, pollIntervalMs, timezoneId, enabled, capabilities, Map.of());
+ }
+
+ /** 读取来源形态的非敏感配置项；缺少该项时返回空值，是否必需由适配器决定。 */
+ public String sourceOption(String key)
+ {
+ return sourceConfig.get(key);
+ }
+
private static String requireText(String value, String field)
{
String normalized = Objects.requireNonNull(value, field).trim();
diff --git a/battery-device-api/src/main/java/com/battery/device/api/registry/DeviceAdapterRegistry.java b/battery-device-api/src/main/java/com/battery/device/api/registry/DeviceAdapterRegistry.java
index e2072aa..f2c51c5 100644
--- a/battery-device-api/src/main/java/com/battery/device/api/registry/DeviceAdapterRegistry.java
+++ b/battery-device-api/src/main/java/com/battery/device/api/registry/DeviceAdapterRegistry.java
@@ -8,6 +8,7 @@
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import com.battery.device.api.DeviceAdapter;
+import com.battery.device.api.TestDataObservationAdapter;
/** 管理启用设备适配器的运行时实例及其会话生命周期。 */
public final class DeviceAdapterRegistry implements AutoCloseable
@@ -34,18 +35,36 @@
return;
}
- RegisteredAdapter previous;
synchronized (this)
{
RegisteredAdapter current = adapters.get(config.adapterCode());
if (current != null && current.config.equals(config)) return;
-
- DeviceAdapterFactory factory = factories.get(normalizeType(config.adapterType()));
- if (factory == null)
+ if (!factories.containsKey(normalizeType(config.adapterType())))
throw new AdapterRegistryException("Unsupported device adapter type: " + config.adapterType());
+ }
- DeviceAdapter replacement = factory.create(config);
- previous = adapters.put(config.adapterCode(), new RegisteredAdapter(config, replacement));
+ DeviceAdapterFactory factory = factories.get(normalizeType(config.adapterType()));
+ register(config, factory.create(config));
+ }
+
+ /**
+ * 注册已构造的运行时实例。代码装配实例与数据库配置实例共用同一份实例表，
+ * 使发现与取数按同一适配器编码解析，避免同编码出现两个实例。
+ */
+ public void register(AdapterInstanceConfig config, DeviceAdapter adapter)
+ {
+ Objects.requireNonNull(config, "config");
+ Objects.requireNonNull(adapter, "adapter");
+ if (!config.enabled())
+ {
+ remove(config.adapterCode());
+ return;
+ }
+
+ RegisteredAdapter previous;
+ synchronized (this)
+ {
+ previous = adapters.put(config.adapterCode(), new RegisteredAdapter(config, adapter));
}
closeWhenIdle(previous);
}
@@ -54,14 +73,7 @@
public <T> T execute(String adapterCode, Function<DeviceAdapter, T> operation)
{
Objects.requireNonNull(operation, "operation");
- RegisteredAdapter registered;
- synchronized (this)
- {
- registered = adapters.get(adapterCode);
- if (registered == null)
- throw new AdapterRegistryException("Device adapter is not enabled: " + adapterCode);
- registered.operationLock.lock();
- }
+ RegisteredAdapter registered = acquire(adapterCode);
try
{
return operation.apply(registered.adapter);
@@ -72,6 +84,41 @@
}
}
+ /**
+ * 在实例级串行边界内按测试数据观测契约执行。
+ * 实例缺失或未提供该契约时明确失败，不回退到其他实例。
+ */
+ public <T> T executeTestData(String adapterCode, Function<TestDataObservationAdapter, T> operation)
+ {
+ Objects.requireNonNull(operation, "operation");
+ RegisteredAdapter registered = acquire(adapterCode);
+ try
+ {
+ if (!(registered.adapter instanceof TestDataObservationAdapter observation))
+ throw new AdapterRegistryException(
+ "Device adapter does not expose test data observation: " + adapterCode);
+ return operation.apply(observation);
+ }
+ finally
+ {
+ registered.operationLock.unlock();
+ }
+ }
+
+ /** 取用实例并持有其串行锁；调用方必须在 finally 中释放。 */
+ private RegisteredAdapter acquire(String adapterCode)
+ {
+ RegisteredAdapter registered;
+ synchronized (this)
+ {
+ registered = adapters.get(adapterCode);
+ if (registered == null)
+ throw new AdapterRegistryException("Device adapter is not enabled: " + adapterCode);
+ registered.operationLock.lock();
+ }
+ return registered;
+ }
+
public synchronized boolean contains(String adapterCode)
{
return adapters.containsKey(adapterCode);
diff --git a/battery-device-api/src/test/java/com/battery/device/api/registry/DeviceAdapterRegistryTest.java b/battery-device-api/src/test/java/com/battery/device/api/registry/DeviceAdapterRegistryTest.java
index 56e6c7e..05e1c91 100644
--- a/battery-device-api/src/test/java/com/battery/device/api/registry/DeviceAdapterRegistryTest.java
+++ b/battery-device-api/src/test/java/com/battery/device/api/registry/DeviceAdapterRegistryTest.java
@@ -1,5 +1,6 @@
package com.battery.device.api.registry;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -15,6 +16,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.jupiter.api.Test;
import com.battery.device.api.DeviceAdapter;
+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.DeviceControlCommand;
@@ -153,6 +155,139 @@
}
}
+ /**
+ * 多来源实例：取数按适配器编码解析到各自实例；实例缺失时明确失败，不回退到其他实例。
+ */
+ @Test
+ void routesTestDataObservationToTheInstanceNamedByTheChannelAndFailsWithoutFallback()
+ {
+ try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of(new ObservableFactory())))
+ {
+ registry.register(observableConfig("source-a", 1000), new ObservableAdapter("source-a", 1000L));
+ registry.register(observableConfig("source-b", 2000), new ObservableAdapter("source-b", 2000L));
+
+ assertEquals(1000L, registry.executeTestData("source-a",
+ adapter -> adapter.latestCabinet(channel("source-a")).sequenceId()).longValue());
+ assertEquals(2000L, registry.executeTestData("source-b",
+ adapter -> adapter.latestCabinet(channel("source-b")).sequenceId()).longValue());
+
+ // 实例缺失或未提供扩展观测契约时明确失败，不能落到别的实例上。
+ assertThrows(AdapterRegistryException.class,
+ () -> registry.executeTestData("source-c", adapter -> adapter.latestCabinet(channel("source-a"))));
+ assertThrows(AdapterRegistryException.class,
+ () -> registry.executeTestData("lab-bts", adapter -> adapter.adapterCode()));
+ }
+ }
+
+ /** 代码装配的实例与工厂实例共用同一份实例表，发现与取数因此按同一编码解析。 */
+ @Test
+ void codeAssembledInstanceIsVisibleToBothAccessors()
+ {
+ try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of()))
+ {
+ ObservableAdapter assembled = new ObservableAdapter("assembled", 42L);
+ registry.register(observableConfig("assembled", 3000), assembled);
+
+ assertSame(assembled, registry.execute("assembled", adapter -> adapter));
+ assertSame(assembled, registry.executeTestData("assembled", adapter -> adapter));
+ }
+ }
+
+ private static ChannelIdentity channel(String adapterCode)
+ {
+ return new ChannelIdentity(adapterCode, 22, 1, 1, 1);
+ }
+
+ private static AdapterInstanceConfig observableConfig(String adapterCode, int pollIntervalMs)
+ {
+ return new AdapterInstanceConfig(adapterCode, "NEWARE_MYSQL", "127.0.0.1", 3306, pollIntervalMs,
+ "Asia/Hong_Kong", true, AdapterCapabilities.observeOnly());
+ }
+
+ private static final class ObservableFactory implements DeviceAdapterFactory
+ {
+ @Override
+ public String adapterType()
+ {
+ return "NEWARE_MYSQL";
+ }
+
+ @Override
+ public DeviceAdapter create(AdapterInstanceConfig config)
+ {
+ return new ObservableAdapter(config.adapterCode(), 0L);
+ }
+ }
+
+ /** 同时实现发现与测试数据观测契约的最小实例，用于验证按编码路由。 */
+ private static final class ObservableAdapter implements DeviceAdapter, TestDataObservationAdapter
+ {
+ private final String code;
+ private final long sequenceId;
+
+ private ObservableAdapter(String code, long sequenceId)
+ {
+ this.code = code;
+ this.sequenceId = sequenceId;
+ }
+
+ @Override
+ public String adapterCode()
+ {
+ return code;
+ }
+
+ @Override
+ public AdapterCapabilities capabilities()
+ {
+ return AdapterCapabilities.observeOnly();
+ }
+
+ @Override
+ public List<com.battery.device.api.model.DetectedChannel> discoverChannels()
+ {
+ return List.of();
+ }
+
+ @Override
+ public java.util.Map<ChannelIdentity, com.battery.device.api.model.RealtimeSnapshot> inquire(
+ java.util.Set<ChannelIdentity> channels)
+ {
+ return java.util.Map.of();
+ }
+
+ @Override
+ public TestDataObservationAdapter.CabinetRealtime latestCabinet(ChannelIdentity channel)
+ {
+ return new TestDataObservationAdapter.CabinetRealtime(sequenceId, java.time.Instant.EPOCH, 0, 0, 0, 0,
+ 0, 0, null, null, java.time.Instant.EPOCH, null, "Asia/Hong_Kong", null);
+ }
+
+ @Override
+ public TestDataObservationAdapter.CollectorRealtime latestCollector(String adapterCode, String collectorCode)
+ {
+ return null;
+ }
+
+ @Override
+ public List<TestDataObservationAdapter.StepRecord> stepRecords(ChannelIdentity channel)
+ {
+ return List.of();
+ }
+
+ @Override
+ public List<TestDataObservationAdapter.CycleRecord> cycleRecords(ChannelIdentity channel)
+ {
+ return List.of();
+ }
+
+ @Override
+ public long persistenceIntervalMillis(ChannelIdentity channel)
+ {
+ return 60000L;
+ }
+ }
+
private static AdapterInstanceConfig config(String host, boolean enabled)
{
return new AdapterInstanceConfig("lab-bts", "BTS", host, 502, 5000,
diff --git a/battery-device-neware/pom.xml b/battery-device-neware/pom.xml
new file mode 100644
index 0000000..c034c22
--- /dev/null
+++ b/battery-device-neware/pom.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent><groupId>com.ruoyi</groupId><artifactId>ruoyi</artifactId><version>3.9.2</version></parent>
+ <groupId>com.battery</groupId><artifactId>battery-device-neware</artifactId>
+ <description>新威上位机数据库的只读直连适配器</description>
+ <dependencies>
+ <dependency><groupId>com.battery</groupId><artifactId>battery-device-api</artifactId></dependency>
+ <dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId></dependency>
+ <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency>
+ </dependencies>
+</project>
diff --git a/battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapter.java b/battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapter.java
new file mode 100644
index 0000000..99c6360
--- /dev/null
+++ b/battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapter.java
@@ -0,0 +1,466 @@
+package com.battery.device.neware;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Clock;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.temporal.ChronoField;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.regex.Pattern;
+import javax.sql.DataSource;
+import com.battery.device.api.DeviceAdapter;
+import com.battery.device.api.DeviceAdapterException;
+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.DetectedDeviceMetadata;
+import com.battery.device.api.model.DeviceConnectionStatus;
+import com.battery.device.api.model.DeviceWorkStatus;
+import com.battery.device.api.model.RealtimeSnapshot;
+import com.battery.device.api.registry.AdapterInstanceConfig;
+
+/**
+ * 直连新威上位机 MySQL 库的只读适配器（存储形态 {@code NEWARE_MYSQL}）。
+ *
+ * <p>现场确认的取数事实（见 {@code docs/commissioning/现场对接清单.md}）：
+ * <ul>
+ * <li>发现候选来自 {@code channel} 通道登记表，主键 {@code (dev_uid, unit_id, chl_id)}；
+ * 设备号 {@code dev_uid = devtype × 10000 + devid}。登记表不等于"当前在用"，
+ * 所以只把"最近数据时间"作为事实交出，登记哪些候选由操作员决定。</li>
+ * <li>逐点数据在 {@code data_<设备六位>_<年>_<月>_<日>} 分表里，表名由 {@code test} 表的
+ * {@code main_first_table} / {@code main_second_table} 直接给出，不按日期推算。</li>
+ * <li>最新测试按 {@code start_time} 取，不能按 {@code test_id} 排序（实测不单调）。</li>
+ * <li>活跃与否只看有没有新的采样行；{@code end_time} 是否为空与 8.0 的通道水位线列都不作为依据。</li>
+ * <li>厂商固定每 60 秒写一行，实时展示节奏与数据产生节奏天生分离。</li>
+ * <li>辅助表族实测没有数据，采集器辅助测量只能走导入，因此 {@link #latestCollector} 保持为空。</li>
+ * </ul>
+ *
+ * <p>单位换算按实测公式：电压与电流除以 10000（原单位 0.1 mV / 0.1 mA），容量与能量除以 36000000
+ * （原单位 1e-4 C / 1e-4 J）。工步类型保留设备给出的数字，不按电流符号猜测方向；
+ * 内阻与温度的单位语义仍未确认，因此不对外暴露。
+ */
+public final class NewareMysqlAdapter implements DeviceAdapter, TestDataObservationAdapter
+{
+ public static final String ADAPTER_TYPE = "NEWARE_MYSQL";
+ /** 连接口令的默认环境变量名；也可由实例配置的 passwordEnv 指定。 */
+ public static final String DEFAULT_PASSWORD_ENV = "BATTERY_VENDOR_DB_PASSWORD";
+ /** 厂商实测写库周期；它同时是持久化周期，展示节奏不随之加快。 */
+ public static final long SAMPLE_INTERVAL_MS = 60000L;
+ private static final int SAMPLE_BATCH_LIMIT = 500;
+ private static final Pattern SAFE_TABLE_NAME = Pattern.compile("[A-Za-z0-9_]{1,64}");
+ /** 设备时间按适配器时区解释；来源列是秒精度 datetime，小数部分按需接受。 */
+ private static final DateTimeFormatter DEVICE_TIME = new DateTimeFormatterBuilder()
+ .appendPattern("yyyy-MM-dd HH:mm:ss")
+ .optionalStart().appendFraction(ChronoField.NANO_OF_SECOND, 1, 9, true).optionalEnd()
+ .toFormatter(Locale.ROOT);
+ /** 水位回写来源时使用秒精度文本，与来源列一致；同秒内的区分由协议序号承担。 */
+ private static final DateTimeFormatter DEVICE_TIME_TEXT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss",
+ Locale.ROOT);
+ private static final BigDecimal VOLTAGE_CURRENT_DIVISOR = BigDecimal.valueOf(10000);
+ private static final BigDecimal CAPACITY_ENERGY_DIVISOR = BigDecimal.valueOf(36000000);
+
+ private final AdapterInstanceConfig config;
+ private final DataSource dataSource;
+ private final Runnable connectionReset;
+ private final Clock clock;
+ private final ZoneId deviceZone;
+ /** 最近数据时间最快也要一分钟才可能变化，缓存使发现清单不必每轮重读全部通道。 */
+ private final long lastDataCacheMillis;
+ private final Map<String, CachedLastData> lastDataCache = new ConcurrentHashMap<>();
+
+ public NewareMysqlAdapter(AdapterInstanceConfig config, DataSource dataSource)
+ {
+ this(config, dataSource, () -> { }, Clock.systemUTC(), SAMPLE_INTERVAL_MS);
+ }
+
+ public NewareMysqlAdapter(AdapterInstanceConfig config, DataSource dataSource, Runnable connectionReset,
+ Clock clock, long lastDataCacheMillis)
+ {
+ this.config = config;
+ this.dataSource = dataSource;
+ this.connectionReset = connectionReset;
+ this.clock = clock;
+ this.deviceZone = ZoneId.of(config.timezoneId());
+ this.lastDataCacheMillis = Math.max(0, lastDataCacheMillis);
+ }
+
+ /** 由工厂使用：把驱动连接包上只读白名单，并在查询失败后作废连接。 */
+ public static NewareMysqlAdapter create(AdapterInstanceConfig config, SingleConnectionVendorDataSource source)
+ {
+ return new NewareMysqlAdapter(config, new ReadOnlyVendorDataSource(source, config.adapterCode()),
+ source::invalidate, Clock.systemUTC(), SAMPLE_INTERVAL_MS);
+ }
+
+ @Override
+ public String adapterCode()
+ {
+ return config.adapterCode();
+ }
+
+ @Override
+ public AdapterCapabilities capabilities()
+ {
+ return config.capabilities();
+ }
+
+ /** 发现候选来自厂商库的通道登记表；每条候选附带该通道最近的采样行时间。 */
+ @Override
+ public List<DetectedChannel> discoverChannels()
+ {
+ List<ChannelKey> registered = query("select dev_uid, unit_id, chl_id from channel "
+ + "order by dev_uid, unit_id, chl_id", this::readRegisteredChannels);
+ Map<Long, List<ChannelKey>> byDevice = new LinkedHashMap<>();
+ for (ChannelKey key : registered)
+ byDevice.computeIfAbsent(key.devUid(), ignored -> new ArrayList<>()).add(key);
+ List<DetectedChannel> channels = new ArrayList<>();
+ Instant observedAt = clock.instant();
+ for (Map.Entry<Long, List<ChannelKey>> entry : byDevice.entrySet())
+ {
+ Map<ChannelKey, VendorTest> latestTests = latestTests(entry.getKey());
+ for (ChannelKey key : entry.getValue())
+ {
+ VendorTest latest = latestTests.get(key);
+ Instant lastDataAt = latest == null ? null : cachedLastSampleAt(key, latest, observedAt);
+ channels.add(new DetectedChannel(identity(key), config.host(), true,
+ DetectedDeviceMetadata.empty(), lastDataAt));
+ }
+ }
+ return List.copyOf(channels);
+ }
+
+ /** 实时快照取自该通道最新测试的最新一行；没有可用行时不返回该通道，避免沿用过期观测。 */
+ @Override
+ public Map<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
+ {
+ if (channels == null || channels.isEmpty()) return Map.of();
+ Map<ChannelIdentity, RealtimeSnapshot> result = new LinkedHashMap<>();
+ for (ChannelIdentity channel : channels)
+ {
+ CabinetRealtime latest = latestCabinet(channel);
+ if (latest == null) continue;
+ result.put(channel, snapshot(channel, latest));
+ }
+ return Map.copyOf(result);
+ }
+
+ @Override
+ public CabinetRealtime latestCabinet(ChannelIdentity channel)
+ {
+ ChannelKey key = channelKey(channel);
+ VendorTest latest = latestTest(key);
+ if (latest == null) return null;
+ SampleRow row = lastSample(key, latest);
+ return row == null ? null : cabinetRealtime(row);
+ }
+
+ /** 新威辅助表族实测没有数据，采集器辅助测量只能靠导入（ADR-0064）。 */
+ @Override
+ public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
+ {
+ return null;
+ }
+
+ /** 工步层记录由本单之后的一张单实现，这里不返回从采样推导出来的工步。 */
+ @Override
+ public List<StepRecord> stepRecords(ChannelIdentity channel)
+ {
+ return List.of();
+ }
+
+ /** 循环汇总表不存在，聚合配方须先用真实数据核对，这里不返回推导值。 */
+ @Override
+ public List<CycleRecord> cycleRecords(ChannelIdentity channel)
+ {
+ return List.of();
+ }
+
+ /** 厂商固定 60 秒写库；持久化周期与之对齐，不做倍速。 */
+ @Override
+ public long persistenceIntervalMillis(ChannelIdentity channel)
+ {
+ return SAMPLE_INTERVAL_MS;
+ }
+
+ /** 水位是来源里已存在的最新一行：设备绝对时间与协议序号。 */
+ @Override
+ public SampleWatermark currentWatermark(ChannelIdentity channel)
+ {
+ ChannelKey key = channelKey(channel);
+ VendorTest latest = latestTest(key);
+ if (latest == null) return new SampleWatermark(null, 0L);
+ SampleRow row = lastSample(key, latest);
+ return row == null ? new SampleWatermark(null, 0L) : new SampleWatermark(row.deviceTime(), row.sequenceId());
+ }
+
+ /**
+ * 严格晚于水位的设备数据行。同一绝对秒内的多行各自成立（实测工步切换点会在同一秒写入多行），
+ * 所以按 {@code (test_atime, seq_id)} 过滤与排序，不按秒合并。
+ *
+ * <p>设备时间为空的水位表示边界时该通道在来源里没有任何行，之后出现的行全部属于本次接收。
+ * 厂商侧测试编号不参与边界判定，只写进来源行身份；接收边界是本系统的接收边界。
+ * 若测试期间分表滚动，两张表都会被读取，避免滚动点整段缺失。
+ */
+ @Override
+ public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
+ {
+ if (watermark == null) return List.of();
+ ChannelKey key = channelKey(channel);
+ VendorTest latest = latestTest(key);
+ if (latest == null) return List.of();
+ boolean bounded = watermark.deviceTime() != null;
+ String rawWatermark = bounded ? deviceTimeText(watermark.deviceTime()) : null;
+ List<SampleRow> rows = new ArrayList<>();
+ for (String table : latest.tables())
+ {
+ String predicate = bounded
+ ? "and (test_atime > ? or (test_atime = ? and seq_id > ?)) "
+ : "";
+ Object[] parameters = bounded
+ ? new Object[] { key.unitId(), key.chlId(), rawWatermark, rawWatermark, watermark.sequenceId() }
+ : new Object[] { key.unitId(), key.chlId() };
+ rows.addAll(query("select test_id, seq_id, cycle, step_id, step_type, test_time, test_vol, test_cur, "
+ + "test_capchg, test_engchg, test_capdchg, test_engdchg, test_atime from " + quote(table)
+ + " where unit_id = ? and chl_id = ? " + predicate
+ + "order by test_atime, seq_id limit " + SAMPLE_BATCH_LIMIT,
+ this::readSampleRows, parameters));
+ }
+ rows.sort(Comparator.comparing(SampleRow::deviceTime).thenComparingLong(SampleRow::sequenceId));
+ return rows.stream().map(this::cabinetRealtime).toList();
+ }
+
+ // ------------------------------------------------------------------ 来源读取
+
+ private List<ChannelKey> readRegisteredChannels(ResultSet rows) throws SQLException
+ {
+ List<ChannelKey> channels = new ArrayList<>();
+ while (rows.next())
+ channels.add(new ChannelKey(rows.getLong("dev_uid"), rows.getInt("unit_id"), rows.getInt("chl_id")));
+ return channels;
+ }
+
+ /** 按起始时间取每台设备每个通道的最新测试；test_id 实测不单调，不能用来排序。 */
+ private Map<ChannelKey, VendorTest> latestTests(long devUid)
+ {
+ return query("select unit_id, chl_id, test_id, main_first_table, main_second_table from test "
+ + "where dev_uid = ? and start_time = (select max(start_time) from test newer "
+ + "where newer.dev_uid = test.dev_uid and newer.unit_id = test.unit_id and newer.chl_id = test.chl_id) "
+ + "order by unit_id, chl_id, test_id", rows -> {
+ Map<ChannelKey, VendorTest> result = new LinkedHashMap<>();
+ while (rows.next())
+ {
+ ChannelKey key = new ChannelKey(devUid, rows.getInt("unit_id"), rows.getInt("chl_id"));
+ result.putIfAbsent(key, vendorTest(rows.getLong("test_id"),
+ rows.getString("main_first_table"), rows.getString("main_second_table")));
+ }
+ return result;
+ }, devUid);
+ }
+
+ private VendorTest latestTest(ChannelKey key)
+ {
+ List<VendorTest> tests = query("select test_id, main_first_table, main_second_table from test "
+ + "where dev_uid = ? and unit_id = ? and chl_id = ? order by start_time desc limit 1",
+ rows -> rows.next() ? List.of(vendorTest(rows.getLong("test_id"), rows.getString("main_first_table"),
+ rows.getString("main_second_table"))) : List.<VendorTest>of(),
+ key.devUid(), key.unitId(), key.chlId());
+ return tests.isEmpty() ? null : tests.get(0);
+ }
+
+ /** 测试数据可能跨分表滚动，两张表都要看；优先取第二张表，滚动后它就是最新的那张。 */
+ private SampleRow lastSample(ChannelKey key, VendorTest test)
+ {
+ List<String> tables = new ArrayList<>(test.tables());
+ Collections.reverse(tables);
+ for (String table : tables)
+ {
+ List<SampleRow> rows = query("select test_id, seq_id, cycle, step_id, step_type, test_time, test_vol, "
+ + "test_cur, test_capchg, test_engchg, test_capdchg, test_engdchg, test_atime from " + quote(table)
+ + " where unit_id = ? and chl_id = ? and test_id = ? order by seq_id desc limit 1",
+ this::readSampleRows, key.unitId(), key.chlId(), test.testId());
+ if (!rows.isEmpty()) return rows.get(0);
+ }
+ return null;
+ }
+
+ /** 最近数据时间来自采样行本身：没有新采样行就是没有，不看结束时间也不看通道水位线列。 */
+ private Instant cachedLastSampleAt(ChannelKey key, VendorTest test, Instant now)
+ {
+ String cacheKey = key.devUid() + ":" + key.unitId() + ":" + key.chlId() + ":" + test.testId();
+ CachedLastData cached = lastDataCache.get(cacheKey);
+ if (cached != null && now.isBefore(cached.expiresAt())) return cached.lastDataAt();
+ SampleRow row = lastSample(key, test);
+ Instant lastDataAt = row == null ? null : row.deviceTime();
+ if (lastDataCacheMillis > 0)
+ lastDataCache.put(cacheKey, new CachedLastData(lastDataAt, now.plusMillis(lastDataCacheMillis)));
+ return lastDataAt;
+ }
+
+ private List<SampleRow> readSampleRows(ResultSet rows) throws SQLException
+ {
+ List<SampleRow> result = new ArrayList<>();
+ while (rows.next())
+ {
+ String raw = rows.getString("test_atime");
+ LocalDateTime deviceTime = parseDeviceTime(raw);
+ result.add(new SampleRow(rows.getLong("test_id"), rows.getLong("seq_id"), rows.getInt("cycle"),
+ rows.getInt("step_id"), rows.getString("step_type"), rows.getLong("test_time"),
+ rows.getBigDecimal("test_vol"), rows.getBigDecimal("test_cur"), rows.getBigDecimal("test_capchg"),
+ rows.getBigDecimal("test_engchg"), rows.getBigDecimal("test_capdchg"),
+ rows.getBigDecimal("test_engdchg"), raw, deviceTime.atZone(deviceZone).toInstant()));
+ }
+ return result;
+ }
+
+ private CabinetRealtime cabinetRealtime(SampleRow row)
+ {
+ // 充放电容量按电流方向取对应的一组：设备界面对正在放电的通道显示放电容量。
+ boolean charging = row.current().signum() >= 0;
+ BigDecimal capacity = charging ? row.capacityCharge() : row.capacityDischarge();
+ BigDecimal energy = charging ? row.energyCharge() : row.energyDischarge();
+ return new CabinetRealtime(row.sequenceId(), row.deviceTime(), scaled(row.current(), VOLTAGE_CURRENT_DIVISOR),
+ scaled(row.voltage(), VOLTAGE_CURRENT_DIVISOR), scaled(capacity, CAPACITY_ENERGY_DIVISOR),
+ scaled(energy, CAPACITY_ENERGY_DIVISOR), row.cycle(), row.stepId(), null, row.stepType(),
+ clock.instant(), row.deviceTimeRaw(), deviceZone.getId(),
+ row.testTimeMillis() == null ? null : row.testTimeMillis() / 1000,
+ "neware:" + row.testId() + ":" + row.sequenceId());
+ }
+
+ private RealtimeSnapshot snapshot(ChannelIdentity channel, CabinetRealtime latest)
+ {
+ // 读库路线下通道工作状态不可观测（ADR-0063）：不按数据行推断，也不沿用上一次状态。
+ return new RealtimeSnapshot(channel, DeviceWorkStatus.UNOBSERVABLE, latest.cycleNo(), latest.stepNo(),
+ latest.stepType(), null, BigDecimal.valueOf(latest.currentA()), BigDecimal.valueOf(latest.voltageV()),
+ BigDecimal.valueOf(latest.capacityAh()), BigDecimal.valueOf(latest.energyWh()), null,
+ latest.relativeTimeSeconds(), null, latest.serverReceivedAt(), Map.of(),
+ DeviceConnectionStatus.ONLINE, latest.deviceTime(), List.of(), null);
+ }
+
+ private static double scaled(BigDecimal value, BigDecimal divisor)
+ {
+ if (value == null) return 0d;
+ // 保留 12 位小数与业务列精度一致，避免在换算环节提前丢失设备原始分辨率。
+ return value.divide(divisor, 12, RoundingMode.HALF_UP).doubleValue();
+ }
+
+ private ChannelIdentity identity(ChannelKey key)
+ {
+ // 设备号编码 devtype × 10000 + devid；unit_id 是该柜内的单元，作为 subdevid 保留。
+ return new ChannelIdentity(config.adapterCode(), (int) (key.devUid() / 10000),
+ (int) (key.devUid() % 10000), key.unitId(), key.chlId());
+ }
+
+ private ChannelKey channelKey(ChannelIdentity channel)
+ {
+ if (!config.adapterCode().equals(channel.adapterCode()))
+ throw new DeviceAdapterException("通道身份不属于本实例：" + channel.adapterCode());
+ return new ChannelKey(channel.devtype() * 10000L + channel.devid(), channel.subdevid(), channel.chlid());
+ }
+
+ private static VendorTest vendorTest(long testId, String firstTable, String secondTable)
+ {
+ List<String> tables = new ArrayList<>(2);
+ if (isSafeTable(firstTable)) tables.add(firstTable.trim());
+ if (isSafeTable(secondTable) && !tables.contains(secondTable.trim())) tables.add(secondTable.trim());
+ return new VendorTest(testId, tables);
+ }
+
+ private static boolean isSafeTable(String table)
+ {
+ return table != null && SAFE_TABLE_NAME.matcher(table.trim()).matches();
+ }
+
+ private static String quote(String table)
+ {
+ // 表名来自厂商库自身的 test 记录，必须先按白名单校验再拼接，避免把库内容当 SQL。
+ if (!isSafeTable(table)) throw new DeviceAdapterException("来源表名不合法：" + table);
+ return "`" + table.trim() + "`";
+ }
+
+ private LocalDateTime parseDeviceTime(String raw)
+ {
+ if (raw == null || raw.isBlank())
+ throw new DeviceAdapterException("PARSE_FAILED", "设备时间为空，无法解释为设备事实", null);
+ String normalized = raw.trim().replace('T', ' ');
+ if (normalized.endsWith("Z")) normalized = normalized.substring(0, normalized.length() - 1).trim();
+ try
+ {
+ return LocalDateTime.parse(normalized, DEVICE_TIME);
+ }
+ catch (RuntimeException exception)
+ {
+ throw new DeviceAdapterException("PARSE_FAILED", "设备时间无法解释：" + raw, exception);
+ }
+ }
+
+ private String deviceTimeText(Instant instant)
+ {
+ return instant.atZone(deviceZone).toLocalDateTime().format(DEVICE_TIME_TEXT);
+ }
+
+ // ------------------------------------------------------------------ JDBC
+
+ /**
+ * 厂商数据源交出的是长期复用的一条连接，这里只关闭语句与结果集，不关闭连接本身；
+ * 查询失败时作废连接，下一轮取数重新建立。
+ */
+ private <T> T query(String sql, ResultSetReader<T> reader, Object... parameters)
+ {
+ try
+ {
+ Connection connection = dataSource.getConnection();
+ try (PreparedStatement statement = connection.prepareStatement(sql))
+ {
+ for (int index = 0; index < parameters.length; index++)
+ statement.setObject(index + 1, parameters[index]);
+ try (ResultSet rows = statement.executeQuery())
+ {
+ return reader.read(rows);
+ }
+ }
+ }
+ catch (NonQueryStatementException exception)
+ {
+ throw exception;
+ }
+ catch (SQLException exception)
+ {
+ connectionReset.run();
+ throw new DeviceAdapterException("CONNECTION_FAILED",
+ "厂商库读取失败：" + config.host() + ":" + config.port() + "，原因：" + exception.getMessage(),
+ exception);
+ }
+ }
+
+ private interface ResultSetReader<T>
+ {
+ T read(ResultSet rows) throws SQLException;
+ }
+
+ private record ChannelKey(long devUid, int unitId, int chlId) { }
+
+ private record VendorTest(long testId, List<String> tables) { }
+
+ private record CachedLastData(Instant lastDataAt, Instant expiresAt) { }
+
+ private record SampleRow(long testId, long sequenceId, int cycle, int stepId, String stepType,
+ Long testTimeMillis, BigDecimal voltage, BigDecimal current, BigDecimal capacityCharge,
+ BigDecimal energyCharge, BigDecimal capacityDischarge, BigDecimal energyDischarge,
+ String deviceTimeRaw, Instant deviceTime) { }
+}
diff --git a/battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapterFactory.java b/battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapterFactory.java
new file mode 100644
index 0000000..34da3c6
--- /dev/null
+++ b/battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapterFactory.java
@@ -0,0 +1,78 @@
+package com.battery.device.neware;
+
+import java.util.Map;
+import com.battery.device.api.DeviceAdapter;
+import com.battery.device.api.DeviceAdapterException;
+import com.battery.device.api.model.AdapterCapabilities;
+import com.battery.device.api.registry.AdapterInstanceConfig;
+import com.battery.device.api.registry.DeviceAdapterFactory;
+
+/**
+ * 按存储形态 {@code NEWARE_MYSQL} 构造直连厂商库的只读适配器实例。
+ *
+ * <p>连接口令只从环境变量或启动参数读取（变量名由实例配置的 {@code passwordEnv} 指定，
+ * 缺省为 {@link NewareMysqlAdapter#DEFAULT_PASSWORD_ENV}），不进数据库也不进仓库。
+ * 账号固定为现场取得的服务端只读账号，全权限账号不进入配置。
+ */
+public final class NewareMysqlAdapterFactory implements DeviceAdapterFactory
+{
+ public static final String DATABASE_OPTION = "database";
+ public static final String USER_OPTION = "user";
+ public static final String PASSWORD_ENV_OPTION = "passwordEnv";
+ private static final String DEFAULT_DATABASE = "bts63";
+ private static final String READ_ONLY_USER = "visitor";
+
+ @Override
+ public String adapterType()
+ {
+ return NewareMysqlAdapter.ADAPTER_TYPE;
+ }
+
+ @Override
+ public DeviceAdapter create(AdapterInstanceConfig config)
+ {
+ String database = option(config.sourceConfig(), DATABASE_OPTION, DEFAULT_DATABASE);
+ String user = option(config.sourceConfig(), USER_OPTION, READ_ONLY_USER);
+ String password = credential(option(config.sourceConfig(), PASSWORD_ENV_OPTION,
+ NewareMysqlAdapter.DEFAULT_PASSWORD_ENV));
+ SingleConnectionVendorDataSource source = new SingleConnectionVendorDataSource(
+ jdbcUrl(config.host(), config.port(), database), user, password);
+ return NewareMysqlAdapter.create(config, source);
+ }
+
+ /** 只读账号口令；启动参数优先于环境变量，便于本机排查而不必改环境。 */
+ static String credential(String variableName)
+ {
+ if (variableName == null || variableName.isBlank())
+ throw new DeviceAdapterException("未配置厂商库口令的取值变量名");
+ String value = System.getenv(variableName);
+ if (value == null || value.isBlank()) value = System.getProperty(variableName);
+ if (value == null || value.isBlank())
+ throw new DeviceAdapterException("缺少厂商库连接口令：请设置环境变量 " + variableName);
+ return value;
+ }
+
+ /**
+ * MySQL 5.5 那台不支持任何会话级只读机制，因此 URL 中不得出现 transaction_read_only
+ * （ADR-0062）；只读由服务端账号与语句白名单两层保证。
+ */
+ static String jdbcUrl(String host, int port, String database)
+ {
+ return "jdbc:mysql://" + host + ":" + port + "/" + database
+ + "?connectTimeout=5000&socketTimeout=10000&useSSL=false&allowPublicKeyRetrieval=true"
+ + "&zeroDateTimeBehavior=convertToNull&connectionTimeZone=Asia/Hong_Kong"
+ + "&forceConnectionTimeZoneToSession=false";
+ }
+
+ private static String option(Map<String, String> sourceConfig, String key, String fallback)
+ {
+ String value = sourceConfig.get(key);
+ return value == null || value.isBlank() ? fallback : value.trim();
+ }
+
+ /** 供装配层声明该形态的运行时能力：可发现、可观测实时值、只读。 */
+ public static AdapterCapabilities capabilities()
+ {
+ return AdapterCapabilities.observeOnly();
+ }
+}
diff --git a/battery-device-neware/src/main/java/com/battery/device/neware/NonQueryStatementException.java b/battery-device-neware/src/main/java/com/battery/device/neware/NonQueryStatementException.java
new file mode 100644
index 0000000..10e114a
--- /dev/null
+++ b/battery-device-neware/src/main/java/com/battery/device/neware/NonQueryStatementException.java
@@ -0,0 +1,12 @@
+package com.battery.device.neware;
+
+/** 非查询语句在到达驱动之前被拒绝；抛出时不会产生任何服务端效果。 */
+public class NonQueryStatementException extends RuntimeException
+{
+ private static final long serialVersionUID = 1L;
+
+ public NonQueryStatementException(String message)
+ {
+ super(message);
+ }
+}
diff --git a/battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyStatementAudit.java b/battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyStatementAudit.java
new file mode 100644
index 0000000..51d69e0
--- /dev/null
+++ b/battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyStatementAudit.java
@@ -0,0 +1,45 @@
+package com.battery.device.neware;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * 厂商数据源上的语句审计：记录已转发到驱动的语句和被白名单挡下的语句。
+ *
+ * <p>它让"该数据源从未发出过非查询语句"成为可断言的事实，而不是纪律性约定；
+ * 断言对应的测试见 {@code ReadOnlyVendorDataSourceTest}。
+ */
+public final class ReadOnlyStatementAudit
+{
+ private final List<String> forwardedStatements = new CopyOnWriteArrayList<>();
+ private final List<String> rejectedStatements = new CopyOnWriteArrayList<>();
+
+ void recordForwarded(String statement)
+ {
+ forwardedStatements.add(abbreviate(statement));
+ }
+
+ void recordRejected(String statement)
+ {
+ rejectedStatements.add(abbreviate(statement));
+ }
+
+ /** 已被转发到驱动的语句。 */
+ public List<String> forwardedStatements()
+ {
+ return List.copyOf(forwardedStatements);
+ }
+
+ /** 被白名单挡下、从未到达驱动的语句。 */
+ public List<String> rejectedStatements()
+ {
+ return List.copyOf(rejectedStatements);
+ }
+
+ private static String abbreviate(String statement)
+ {
+ if (statement == null) return "";
+ String trimmed = statement.trim();
+ return trimmed.length() <= 200 ? trimmed : trimmed.substring(0, 200);
+ }
+}
diff --git a/battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyVendorDataSource.java b/battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyVendorDataSource.java
new file mode 100644
index 0000000..8562e73
--- /dev/null
+++ b/battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyVendorDataSource.java
@@ -0,0 +1,237 @@
+package com.battery.device.neware;
+
+import java.io.PrintWriter;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.sql.CallableStatement;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Locale;
+import java.util.logging.Logger;
+import javax.sql.DataSource;
+
+/**
+ * 厂商数据源的只读防线：在 JDBC 调用到达驱动之前按语句白名单放行。
+ *
+ * <p>只允许 {@code SELECT} / {@code SHOW} / {@code DESCRIBE}（含 {@code DESC}）；
+ * 其余语句（含存储过程调用）直接抛 {@link NonQueryStatementException}，不转发给驱动，
+ * 因此在服务端不产生任何效果。被拒绝的语句记入 {@link ReadOnlyStatementAudit}，
+ * 使"该数据源从未发出过非查询语句"可以被测试断言。
+ *
+ * <p>防线是结构性的，不是数学上的：任何绕过本数据源另开连接的代码仍能写。
+ * 服务端只读账号（见现场对接清单第七节）是第一道防线，本层是第二道。
+ */
+public final class ReadOnlyVendorDataSource implements DataSource
+{
+ private static final List<String> ALLOWED_PREFIXES = List.of("select", "show", "describe", "desc");
+
+ private final DataSource delegate;
+ private final String sourceLabel;
+ private final ReadOnlyStatementAudit audit = new ReadOnlyStatementAudit();
+
+ public ReadOnlyVendorDataSource(DataSource delegate, String sourceLabel)
+ {
+ if (delegate == null) throw new IllegalArgumentException("delegate data source must not be null");
+ this.delegate = delegate;
+ this.sourceLabel = sourceLabel == null || sourceLabel.isBlank() ? "vendor" : sourceLabel.trim();
+ }
+
+ /** 返回本数据源上的语句审计，用于断言从未发出过非查询语句。 */
+ public ReadOnlyStatementAudit audit()
+ {
+ return audit;
+ }
+
+ @Override
+ public Connection getConnection() throws SQLException
+ {
+ return guard(delegate.getConnection());
+ }
+
+ @Override
+ public Connection getConnection(String username, String password) throws SQLException
+ {
+ return guard(delegate.getConnection(username, password));
+ }
+
+ private Connection guard(Connection connection)
+ {
+ return (Connection) Proxy.newProxyInstance(connection.getClass().getClassLoader(),
+ new Class<?>[] { Connection.class }, (proxy, method, args) -> {
+ switch (method.getName())
+ {
+ case "prepareStatement" ->
+ {
+ String sql = (String) args[0];
+ requireQuery(sql);
+ return guardStatement((Statement) invoke(connection, method, args), sql, true);
+ }
+ case "prepareCall" -> throw new NonQueryStatementException(
+ "厂商数据源禁止存储过程调用：" + abbreviate((String) args[0]));
+ case "createStatement" ->
+ {
+ return guardStatement((Statement) invoke(connection, method, args), null, false);
+ }
+ default -> { }
+ }
+ return invoke(connection, method, args);
+ });
+ }
+
+ private Statement guardStatement(Statement statement, String preparedSql, boolean prepared)
+ {
+ Class<?> statementType = prepared ? PreparedStatement.class : Statement.class;
+ return (Statement) Proxy.newProxyInstance(statement.getClass().getClassLoader(),
+ new Class<?>[] { statementType }, (proxy, method, args) -> {
+ String sql = sqlArgument(method, args);
+ if (sql != null) requireQuery(sql);
+ return invoke(statement, method, args);
+ });
+ }
+
+ /**
+ * 取出本次调用携带的 SQL 文本：语句级方法带 SQL 时按该 SQL 判定，
+ * 预编译语句的无参执行按 prepare 时的 SQL 判定（已在 prepareStatement 中放行）。
+ */
+ private static String sqlArgument(Method method, Object[] args)
+ {
+ if (args == null || args.length == 0 || !(args[0] instanceof String sql)) return null;
+ return switch (method.getName())
+ {
+ case "execute", "executeUpdate", "executeLargeUpdate", "executeQuery", "addBatch", "executeBatch",
+ "executeLargeBatch" -> sql;
+ default -> null;
+ };
+ }
+
+ private void requireQuery(String sql)
+ {
+ if (isQuery(sql))
+ {
+ audit.recordForwarded(sql);
+ return;
+ }
+ audit.recordRejected(sql);
+ throw new NonQueryStatementException("厂商数据源只允许查询语句：" + abbreviate(sql));
+ }
+
+ /** 只按语句首关键字判定，避免用正则近似 SQL 语法。 */
+ static boolean isQuery(String sql)
+ {
+ if (sql == null) return false;
+ String normalized = stripLeadingComments(sql).toLowerCase(Locale.ROOT);
+ if (normalized.isEmpty()) return false;
+ for (String prefix : ALLOWED_PREFIXES)
+ {
+ if (!normalized.startsWith(prefix)) continue;
+ int boundary = prefix.length();
+ if (normalized.length() == boundary || Character.isWhitespace(normalized.charAt(boundary))
+ || normalized.charAt(boundary) == '(' || normalized.charAt(boundary) == '*'
+ || normalized.charAt(boundary) == '`')
+ return true;
+ }
+ return false;
+ }
+
+ private static String stripLeadingComments(String sql)
+ {
+ String remaining = sql.trim();
+ while (true)
+ {
+ if (remaining.startsWith("/*"))
+ {
+ int end = remaining.indexOf("*/");
+ if (end < 0) return remaining;
+ remaining = remaining.substring(end + 2).trim();
+ continue;
+ }
+ if (remaining.startsWith("--") || remaining.startsWith("#"))
+ {
+ int end = remaining.indexOf('\n');
+ if (end < 0) return "";
+ remaining = remaining.substring(end + 1).trim();
+ continue;
+ }
+ return remaining;
+ }
+ }
+
+ private static String abbreviate(String sql)
+ {
+ if (sql == null) return "";
+ String trimmed = sql.trim();
+ return trimmed.length() <= 200 ? trimmed : trimmed.substring(0, 200);
+ }
+
+ private static Object invoke(Object target, Method method, Object[] args) throws Throwable
+ {
+ try
+ {
+ return method.invoke(target, args);
+ }
+ catch (InvocationTargetException exception)
+ {
+ throw exception.getTargetException();
+ }
+ }
+
+ @Override
+ public PrintWriter getLogWriter() throws SQLException
+ {
+ return delegate.getLogWriter();
+ }
+
+ @Override
+ public void setLogWriter(PrintWriter out) throws SQLException
+ {
+ delegate.setLogWriter(out);
+ }
+
+ @Override
+ public void setLoginTimeout(int seconds) throws SQLException
+ {
+ delegate.setLoginTimeout(seconds);
+ }
+
+ @Override
+ public int getLoginTimeout() throws SQLException
+ {
+ return delegate.getLoginTimeout();
+ }
+
+ @Override
+ public Logger getParentLogger() throws SQLFeatureNotSupportedException
+ {
+ return delegate.getParentLogger();
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException
+ {
+ if (iface.isInstance(this)) return iface.cast(this);
+ throw new SQLFeatureNotSupportedException("厂商数据源不暴露未受约束的驱动对象");
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface)
+ {
+ return iface.isInstance(this);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "ReadOnlyVendorDataSource[" + sourceLabel + "]";
+ }
+
+ /** 便于测试与排查的来源说明。 */
+ public String sourceLabel()
+ {
+ return sourceLabel;
+ }
+}
diff --git a/battery-device-neware/src/main/java/com/battery/device/neware/SingleConnectionVendorDataSource.java b/battery-device-neware/src/main/java/com/battery/device/neware/SingleConnectionVendorDataSource.java
new file mode 100644
index 0000000..9c03559
--- /dev/null
+++ b/battery-device-neware/src/main/java/com/battery/device/neware/SingleConnectionVendorDataSource.java
@@ -0,0 +1,110 @@
+package com.battery.device.neware;
+
+import java.io.PrintWriter;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.Properties;
+import java.util.logging.Logger;
+import javax.sql.DataSource;
+
+/**
+ * 由 DriverManager 直连厂商库的极简数据源：只被设备适配器使用，不注册进 Spring，
+ * 因此不参与任何业务事务，Druid 主数据源的回滚不会波及厂商库。
+ *
+ * <p>连接复用一条并在查询失败时作废，避免厂商服务端重启后长期持有失效连接；
+ * 厂商库实测按固定 60 秒写库，取数频率远低于连接复用带来的收益门槛。
+ */
+public final class SingleConnectionVendorDataSource implements DataSource
+{
+ private final String url;
+ private final Properties credentials;
+ private Connection connection;
+
+ public SingleConnectionVendorDataSource(String url, String user, String password)
+ {
+ if (url == null || url.isBlank()) throw new IllegalArgumentException("vendor jdbc url must not be blank");
+ this.url = url;
+ this.credentials = new Properties();
+ this.credentials.setProperty("user", user == null ? "" : user);
+ this.credentials.setProperty("password", password == null ? "" : password);
+ }
+
+ @Override
+ public synchronized Connection getConnection() throws SQLException
+ {
+ if (connection == null || connection.isClosed())
+ {
+ connection = DriverManager.getConnection(url, credentials);
+ }
+ return connection;
+ }
+
+ @Override
+ public Connection getConnection(String username, String password) throws SQLException
+ {
+ return DriverManager.getConnection(url, username, password);
+ }
+
+ /** 作废当前连接，使下一次取数重新建立；用于查询失败后的自愈。 */
+ public synchronized void invalidate()
+ {
+ if (connection == null) return;
+ try
+ {
+ connection.close();
+ }
+ catch (SQLException ignored)
+ {
+ // 连接已失效，无需处理。
+ }
+ finally
+ {
+ connection = null;
+ }
+ }
+
+ @Override
+ public PrintWriter getLogWriter() throws SQLException
+ {
+ return DriverManager.getLogWriter();
+ }
+
+ @Override
+ public void setLogWriter(PrintWriter out) throws SQLException
+ {
+ DriverManager.setLogWriter(out);
+ }
+
+ @Override
+ public void setLoginTimeout(int seconds) throws SQLException
+ {
+ DriverManager.setLoginTimeout(seconds);
+ }
+
+ @Override
+ public int getLoginTimeout() throws SQLException
+ {
+ return DriverManager.getLoginTimeout();
+ }
+
+ @Override
+ public Logger getParentLogger() throws SQLFeatureNotSupportedException
+ {
+ return Logger.getLogger("com.battery.device.neware");
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException
+ {
+ if (iface.isInstance(this)) return iface.cast(this);
+ throw new SQLFeatureNotSupportedException("不支持解包为 " + iface);
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface)
+ {
+ return iface.isInstance(this);
+ }
+}
diff --git a/battery-device-neware/src/test/java/com/battery/device/neware/FakeVendorDatabase.java b/battery-device-neware/src/test/java/com/battery/device/neware/FakeVendorDatabase.java
new file mode 100644
index 0000000..62f1a60
--- /dev/null
+++ b/battery-device-neware/src/test/java/com/battery/device/neware/FakeVendorDatabase.java
@@ -0,0 +1,333 @@
+package com.battery.device.neware;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import javax.sql.DataSource;
+
+/**
+ * 可控的假厂商库：按新威 7.6 实测表结构（{@code channel} / {@code test} / {@code data_*}）回答适配器的查询，
+ * 并把每一条到达"驱动"的语句原样记录下来。
+ *
+ * <p>它只回答适配器实际使用的查询形态，因此测试断言的是外部行为（发现清单、最新测试、分表定位、
+ * 单位换算、水位过滤），不是 SQL 文本本身；记录下来的语句用于断言只读防线没有漏过非查询语句。
+ */
+final class FakeVendorDatabase
+{
+ record ChannelRow(long devUid, int unitId, int chlId) { }
+
+ record TestRow(long devUid, int unitId, int chlId, long testId, String startTime,
+ String firstTable, String secondTable) { }
+
+ record SampleRow(String table, long testId, int unitId, int chlId, long seqId, int cycle, int stepId,
+ String stepType, long testTimeMillis, int voltage, int current, int capacityCharge, int energyCharge,
+ int capacityDischarge, int energyDischarge, String testAtime) { }
+
+ private final List<String> statements = new java.util.concurrent.CopyOnWriteArrayList<>();
+ private final List<ChannelRow> channels = new ArrayList<>();
+ private final List<TestRow> tests = new ArrayList<>();
+ private final List<SampleRow> samples = new ArrayList<>();
+
+ FakeVendorDatabase registeredChannel(long devUid, int unitId, int chlId)
+ {
+ channels.add(new ChannelRow(devUid, unitId, chlId));
+ return this;
+ }
+
+ FakeVendorDatabase test(TestRow row)
+ {
+ tests.add(row);
+ return this;
+ }
+
+ FakeVendorDatabase sample(SampleRow row)
+ {
+ samples.add(row);
+ return this;
+ }
+
+ /** 到达"驱动"的语句原文，按调用顺序。 */
+ List<String> statements()
+ {
+ return List.copyOf(statements);
+ }
+
+ DataSource dataSource()
+ {
+ return new DataSource()
+ {
+ @Override
+ public Connection getConnection()
+ {
+ return connection();
+ }
+
+ @Override
+ public Connection getConnection(String username, String password)
+ {
+ return connection();
+ }
+
+ @Override
+ public java.io.PrintWriter getLogWriter()
+ {
+ return null;
+ }
+
+ @Override
+ public void setLogWriter(java.io.PrintWriter out)
+ {
+ }
+
+ @Override
+ public void setLoginTimeout(int seconds)
+ {
+ }
+
+ @Override
+ public int getLoginTimeout()
+ {
+ return 0;
+ }
+
+ @Override
+ public java.util.logging.Logger getParentLogger()
+ {
+ return java.util.logging.Logger.getLogger("fake-vendor");
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException
+ {
+ throw new SQLException("不支持解包");
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface)
+ {
+ return false;
+ }
+ };
+ }
+
+ private Connection connection()
+ {
+ return (Connection) Proxy.newProxyInstance(getClass().getClassLoader(), new Class<?>[] { Connection.class },
+ (proxy, method, args) -> {
+ if ("prepareStatement".equals(method.getName())) return statement((String) args[0]);
+ if ("createStatement".equals(method.getName())) return plainStatement();
+ return defaultValue(method);
+ });
+ }
+
+ /** 语句级执行：SQL 在执行时才出现，因此在这里才记录；探测类语句不回数据。 */
+ private java.sql.Statement plainStatement()
+ {
+ return (java.sql.Statement) Proxy.newProxyInstance(getClass().getClassLoader(),
+ new Class<?>[] { java.sql.Statement.class }, (proxy, method, args) -> {
+ if ("executeQuery".equals(method.getName()) || "execute".equals(method.getName())
+ || "executeUpdate".equals(method.getName()))
+ {
+ statements.add((String) args[0]);
+ return resultSet(List.of());
+ }
+ return defaultValue(method);
+ });
+ }
+
+ private PreparedStatement statement(String sql)
+ {
+ statements.add(sql);
+ List<Object> parameters = new ArrayList<>();
+ return (PreparedStatement) Proxy.newProxyInstance(getClass().getClassLoader(),
+ new Class<?>[] { PreparedStatement.class }, (proxy, method, args) -> {
+ switch (method.getName())
+ {
+ case "setObject", "setString", "setLong", "setInt" ->
+ {
+ int index = (Integer) args[0];
+ while (parameters.size() < index) parameters.add(null);
+ parameters.set(index - 1, args[1]);
+ return null;
+ }
+ case "executeQuery" -> { return resultSet(resolve(sql, parameters)); }
+ case "close" -> { return null; }
+ default -> { return defaultValue(method); }
+ }
+ });
+ }
+
+ /** 按适配器使用的查询形态分派；参数顺序与适配器绑定顺序一致。 */
+ private List<Map<String, Object>> resolve(String sql, List<Object> parameters)
+ {
+ String normalized = sql.toLowerCase().replaceAll("\\s+", " ");
+ if (normalized.contains("from channel"))
+ return channels.stream().sorted(Comparator.comparingLong(ChannelRow::devUid)
+ .thenComparingInt(ChannelRow::unitId).thenComparingInt(ChannelRow::chlId))
+ .map(row -> {
+ Map<String, Object> value = new LinkedHashMap<>();
+ value.put("dev_uid", row.devUid());
+ value.put("unit_id", row.unitId());
+ value.put("chl_id", row.chlId());
+ return value;
+ })
+ .toList();
+ if (normalized.contains("from test") && normalized.contains("max(start_time)"))
+ return latestTests(longValue(parameters, 0));
+ if (normalized.contains("from test"))
+ return latestTestForChannel(longValue(parameters, 0), intValue(parameters, 1), intValue(parameters, 2));
+ if (normalized.contains("order by test_atime, seq_id limit"))
+ return samplesAfter(tableOf(sql), parameters);
+ return lastSampleOfTest(tableOf(sql), intValue(parameters, 0), intValue(parameters, 1),
+ longValue(parameters, 2));
+ }
+
+ private List<Map<String, Object>> latestTests(long devUid)
+ {
+ Map<String, TestRow> latest = new LinkedHashMap<>();
+ tests.stream().filter(row -> row.devUid() == devUid)
+ .sorted(Comparator.comparing(TestRow::startTime).reversed().thenComparingLong(TestRow::testId))
+ .forEach(row -> latest.putIfAbsent(row.unitId() + ":" + row.chlId(), row));
+ return latest.values().stream().sorted(Comparator.comparingInt(TestRow::unitId)
+ .thenComparingInt(TestRow::chlId)).map(this::testColumns).toList();
+ }
+
+ private List<Map<String, Object>> latestTestForChannel(long devUid, int unitId, int chlId)
+ {
+ return tests.stream().filter(row -> row.devUid() == devUid && row.unitId() == unitId && row.chlId() == chlId)
+ .sorted(Comparator.comparing(TestRow::startTime).reversed().thenComparingLong(TestRow::testId))
+ .limit(1).map(this::testColumns).toList();
+ }
+
+ private List<Map<String, Object>> lastSampleOfTest(String table, int unitId, int chlId, long testId)
+ {
+ return samples.stream().filter(row -> row.table().equals(table) && row.unitId() == unitId
+ && row.chlId() == chlId && row.testId() == testId)
+ .sorted(Comparator.comparingLong(SampleRow::seqId).reversed()).limit(1)
+ .map(this::sampleColumns).toList();
+ }
+
+ private List<Map<String, Object>> samplesAfter(String table, List<Object> parameters)
+ {
+ int unitId = intValue(parameters, 0), chlId = intValue(parameters, 1);
+ // 没有带时间的下界时（水位表示界时来源里没有行）返回该通道的全部行。
+ boolean bounded = parameters.size() >= 5;
+ String watermark = bounded ? String.valueOf(parameters.get(2)) : null;
+ long watermarkSeq = bounded ? longValue(parameters, 4) : 0L;
+ return samples.stream().filter(row -> row.table().equals(table) && row.unitId() == unitId
+ && row.chlId() == chlId
+ && (!bounded || after(row.testAtime(), row.seqId(), watermark, watermarkSeq)))
+ .sorted(Comparator.comparing(SampleRow::testAtime).thenComparingLong(SampleRow::seqId))
+ .map(this::sampleColumns).toList();
+ }
+
+ private static boolean after(String atime, long seqId, String watermark, long watermarkSeq)
+ {
+ int time = atime.compareTo(watermark);
+ return time > 0 || (time == 0 && seqId > watermarkSeq);
+ }
+
+ private static String tableOf(String sql)
+ {
+ java.util.regex.Matcher matcher = java.util.regex.Pattern.compile("from\\s+`([A-Za-z0-9_]+)`")
+ .matcher(sql.toLowerCase());
+ if (!matcher.find()) throw new IllegalStateException("假厂商库无法识别分表名：" + sql);
+ return matcher.group(1);
+ }
+
+ private Map<String, Object> testColumns(TestRow row)
+ {
+ Map<String, Object> value = new LinkedHashMap<>();
+ value.put("unit_id", row.unitId());
+ value.put("chl_id", row.chlId());
+ value.put("test_id", row.testId());
+ value.put("main_first_table", row.firstTable());
+ value.put("main_second_table", row.secondTable());
+ return value;
+ }
+
+ private Map<String, Object> sampleColumns(SampleRow row)
+ {
+ Map<String, Object> value = new LinkedHashMap<>();
+ value.put("test_id", row.testId());
+ value.put("seq_id", row.seqId());
+ value.put("cycle", row.cycle());
+ value.put("step_id", row.stepId());
+ value.put("step_type", row.stepType());
+ value.put("test_time", row.testTimeMillis());
+ value.put("test_vol", row.voltage());
+ value.put("test_cur", row.current());
+ value.put("test_capchg", row.capacityCharge());
+ value.put("test_engchg", row.energyCharge());
+ value.put("test_capdchg", row.capacityDischarge());
+ value.put("test_engdchg", row.energyDischarge());
+ value.put("test_atime", row.testAtime());
+ return value;
+ }
+
+ private static ResultSet resultSet(List<Map<String, Object>> rows)
+ {
+ List<String> labels = rows.isEmpty() ? List.of() : List.copyOf(rows.get(0).keySet());
+ int[] cursor = { -1 };
+ return (ResultSet) Proxy.newProxyInstance(FakeVendorDatabase.class.getClassLoader(),
+ new Class<?>[] { ResultSet.class }, (proxy, method, args) -> {
+ switch (method.getName())
+ {
+ case "next" -> { cursor[0]++; return cursor[0] < rows.size(); }
+ case "close" -> { return null; }
+ case "getLong", "getInt", "getString", "getBigDecimal" ->
+ {
+ Object value = value(rows.get(cursor[0]), labels, args[0]);
+ if (value == null) return null;
+ return switch (method.getName())
+ {
+ case "getLong" -> ((Number) value).longValue();
+ case "getInt" -> ((Number) value).intValue();
+ case "getString" -> String.valueOf(value);
+ default -> value instanceof BigDecimal decimal ? decimal
+ : BigDecimal.valueOf(((Number) value).longValue());
+ };
+ }
+ default -> { return defaultValue(method); }
+ }
+ });
+ }
+
+ private static Object value(Map<String, Object> row, List<String> labels, Object key)
+ {
+ if (key instanceof Integer index) return row.get(labels.get(index - 1));
+ return row.get(((String) key).toLowerCase());
+ }
+
+ private static int intValue(List<Object> parameters, int index)
+ {
+ return ((Number) parameters.get(index)).intValue();
+ }
+
+ private static long longValue(List<Object> parameters, int index)
+ {
+ return ((Number) parameters.get(index)).longValue();
+ }
+
+ private static Object defaultValue(Method method)
+ {
+ Class<?> type = method.getReturnType();
+ if (!type.isPrimitive() || type == void.class) return null;
+ if (type == boolean.class) return false;
+ if (type == long.class) return 0L;
+ if (type == double.class) return 0d;
+ if (type == float.class) return 0f;
+ if (type == char.class) return (char) 0;
+ return 0;
+ }
+}
diff --git a/battery-device-neware/src/test/java/com/battery/device/neware/NewareMysqlAdapterContractTest.java b/battery-device-neware/src/test/java/com/battery/device/neware/NewareMysqlAdapterContractTest.java
new file mode 100644
index 0000000..d89ff23
--- /dev/null
+++ b/battery-device-neware/src/test/java/com/battery/device/neware/NewareMysqlAdapterContractTest.java
@@ -0,0 +1,207 @@
+package com.battery.device.neware;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
+import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
+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.DeviceConnectionStatus;
+import com.battery.device.api.model.DeviceWorkStatus;
+import com.battery.device.api.model.RealtimeSnapshot;
+import com.battery.device.api.registry.AdapterInstanceConfig;
+
+/**
+ * 直连新威库的适配器契约测试：用可控的假厂商库验证发现清单与最近数据时间、
+ * 最新测试按起始时间取、分表定位用测试记录给出的表名、活跃只看新采样行、单位换算、身份与水位。
+ */
+class NewareMysqlAdapterContractTest
+{
+ private static final String ADAPTER_CODE = "neware76";
+ private static final long DEV_UID = 210036L;
+ private static final ChannelIdentity CHANNEL = new ChannelIdentity(ADAPTER_CODE, 21, 36, 1, 2);
+ private static final Clock CLOCK = Clock.fixed(Instant.parse("2026-09-28T02:00:00Z"), ZoneOffset.UTC);
+
+ private static final String ACTIVE_TABLE = "data_210036_2026_09_18";
+
+ private NewareMysqlAdapter adapter(FakeVendorDatabase database)
+ {
+ AdapterInstanceConfig config = new AdapterInstanceConfig(ADAPTER_CODE,
+ NewareMysqlAdapter.ADAPTER_TYPE, "26.1.2.3", 3306, 60000, "Asia/Hong_Kong", true,
+ AdapterCapabilities.observeOnly(), Map.of("database", "bts63", "user", "visitor"));
+ // 缓存关闭，让每个断言都读到假厂商库的当前内容。
+ return new NewareMysqlAdapter(config, database.dataSource(), () -> { }, CLOCK, 0);
+ }
+
+ /** 一台在用设备（dev_uid 210036）、一台只出现在历史数据里但不在登记表中的设备（240098）。 */
+ private FakeVendorDatabase activeDeviceDatabase()
+ {
+ return new FakeVendorDatabase()
+ .registeredChannel(DEV_UID, 1, 1)
+ .registeredChannel(DEV_UID, 1, 2)
+ .registeredChannel(DEV_UID, 2, 1)
+ .test(new FakeVendorDatabase.TestRow(DEV_UID, 1, 2, 440L, "2026-09-18 09:00:00",
+ ACTIVE_TABLE, null))
+ // 已退役设备的历史数据：登记表里没有它，不得成为候选。
+ .test(new FakeVendorDatabase.TestRow(240098L, 1, 1, 900L, "2021-09-30 09:00:00",
+ "data_240098_2021_09_30", null))
+ .sample(new FakeVendorDatabase.SampleRow(ACTIVE_TABLE, 440L, 1, 2, 7435L, 3, 12, "2", 300000L,
+ 13733, -11889, 50350024, 1103331, 50350024, 1103331, "2026-09-28 09:54:56"))
+ .sample(new FakeVendorDatabase.SampleRow(ACTIVE_TABLE, 440L, 1, 2, 7436L, 3, 9, "4", 0L,
+ 13700, 0, 0, 0, 0, 0, "2026-09-28 09:54:56"));
+ }
+
+ @Test
+ void discoveryListsRegisteredChannelsWithLastDataTimeOnlyFromSampleRows()
+ {
+ List<DetectedChannel> channels = adapter(activeDeviceDatabase()).discoverChannels();
+ assertEquals(3, channels.size());
+ // 候选集合来自通道登记表：只出现在历史数据里的退役设备不出现。
+ assertTrue(channels.stream().allMatch(item -> item.identity().devid() == 36));
+ DetectedChannel active = channels.stream()
+ .filter(item -> item.identity().chlid() == 2).findFirst().orElseThrow();
+ assertEquals(new ChannelIdentity(ADAPTER_CODE, 21, 36, 1, 2), active.identity());
+ assertTrue(active.available());
+ // 最近数据时间取该通道最新采样行的设备时间（按适配器时区解释）：Asia/Hong_Kong 09:54:56 = 01:54:56Z。
+ assertEquals(Instant.parse("2026-09-28T01:54:56Z"), active.lastDataAt());
+ // 没有采样行的通道不报最近数据时间：活跃只看有没有新采样行写入。
+ DetectedChannel idle = channels.stream()
+ .filter(item -> item.identity().chlid() == 1 && item.identity().subdevid() == 1)
+ .findFirst().orElseThrow();
+ assertNull(idle.lastDataAt());
+ }
+
+ @Test
+ void latestTestIsPickedByStartTimeNotByTestId()
+ {
+ FakeVendorDatabase database = new FakeVendorDatabase()
+ .registeredChannel(DEV_UID, 1, 2)
+ // 测试编号更大的那条起始时间更早：按 test_id 排序会取错。
+ .test(new FakeVendorDatabase.TestRow(DEV_UID, 1, 2, 2818575907L, "2026-02-20 09:00:00",
+ "data_210036_2026_02_20", null))
+ .test(new FakeVendorDatabase.TestRow(DEV_UID, 1, 2, 2818575540L, "2026-09-28 09:00:00",
+ "data_210036_2026_09_28", null))
+ .sample(new FakeVendorDatabase.SampleRow("data_210036_2026_02_20", 2818575907L, 1, 2, 10L, 1, 1, "1",
+ 1000L, 5000, 1000, 10, 10, 10, 10, "2026-02-20 09:00:01"))
+ .sample(new FakeVendorDatabase.SampleRow("data_210036_2026_09_28", 2818575540L, 1, 2, 20L, 4, 7, "2",
+ 60000L, 31541, -12000, 700000, 900000, 800000, 1000000, "2026-09-28 09:01:00"));
+ CabinetRealtime latest = adapter(database).latestCabinet(CHANNEL);
+ assertEquals(20L, latest.sequenceId());
+ assertEquals(Instant.parse("2026-09-28T01:01:00Z"), latest.deviceTime());
+ }
+
+ @Test
+ void shardTableComesFromTheTestRecordInsteadOfTheDeviceTimeDate()
+ {
+ // 实测表名按建表日：9-28 启动的测试写进 9-18 的表。适配器必须用测试记录给出的表名。
+ CabinetRealtime latest = adapter(activeDeviceDatabase()).latestCabinet(CHANNEL);
+ assertEquals("2026-09-28 09:54:56", latest.deviceTimeRaw());
+ assertEquals(Instant.parse("2026-09-28T01:54:56Z"), latest.deviceTime());
+ assertEquals("Asia/Hong_Kong", latest.timezoneId());
+ assertEquals(CLOCK.instant(), latest.serverReceivedAt());
+ }
+
+ @Test
+ void sampledValuesAreConvertedWithTheFieldMeasuredFormula()
+ {
+ // 同秒两行的第一行承载本次实测的换算样本。
+ List<CabinetRealtime> rows = adapter(activeDeviceDatabase()).samplesSince(CHANNEL,
+ new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7434L));
+ CabinetRealtime latest = rows.get(0);
+ // test_vol 13733 → 1.3733 V；test_cur -11889 → -1.1889 A；放电容量与能量除以 36000000。
+ assertEquals(1.3733, latest.voltageV(), 1e-9);
+ assertEquals(-1.1889, latest.currentA(), 1e-9);
+ assertEquals(50350024d / 36000000d, latest.capacityAh(), 1e-9);
+ assertEquals(1103331d / 36000000d, latest.energyWh(), 1e-9);
+ // test_time 单位为毫秒，作为工步内相对时长输出秒。
+ assertEquals(300L, latest.relativeTimeSeconds());
+ assertEquals(3, latest.cycleNo());
+ assertEquals(12, latest.stepNo());
+ assertEquals("2", latest.stepType());
+ assertEquals("neware:440:7435", latest.sourceRecordId());
+ // 工步类型保留设备给出的数字，不按电流符号猜方向；同一秒的另一行靠工步与序号区分。
+ assertEquals(9, rows.get(1).stepNo());
+ }
+
+ @Test
+ void onlineFactIsSourceReachabilityAndChannelWorkStatusStaysUnobservable()
+ {
+ NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
+ Map<ChannelIdentity, RealtimeSnapshot> snapshots = adapter.inquire(Set.of(CHANNEL));
+ RealtimeSnapshot snapshot = snapshots.get(CHANNEL);
+ assertEquals(DeviceConnectionStatus.ONLINE, snapshot.connectionStatus());
+ // 读库路线下通道工作状态不可观测，不按数据行推断。
+ assertEquals(DeviceWorkStatus.UNOBSERVABLE, snapshot.workStatus());
+ assertFalse(adapter.capabilities().control());
+ // 辅助测量来自导入，不来自新威辅助表族。
+ assertNull(adapter.latestCollector(ADAPTER_CODE, "anything"));
+ assertEquals(NewareMysqlAdapter.SAMPLE_INTERVAL_MS, adapter.persistenceIntervalMillis(CHANNEL));
+ }
+
+ @Test
+ void samplesSinceKeepsEveryRowInTheSameSecondAndDropsRowsAtOrBeforeTheWatermark()
+ {
+ NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
+ SampleWatermark watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7434L);
+ List<CabinetRealtime> rows = adapter.samplesSince(CHANNEL, watermark);
+ // 同一绝对秒内的两行都保留，并且靠协议序号区分。
+ assertEquals(2, rows.size());
+ assertEquals(List.of(7435L, 7436L), rows.stream().map(CabinetRealtime::sequenceId).toList());
+ assertEquals(List.of("neware:440:7435", "neware:440:7436"),
+ rows.stream().map(CabinetRealtime::sourceRecordId).toList());
+ // 严格大于水位：等于水位的行不返回。
+ assertEquals(1, adapter.samplesSince(CHANNEL,
+ new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7435L)).size());
+ // 水位之后没有新行时返回空。
+ assertTrue(adapter.samplesSince(CHANNEL,
+ new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L)).isEmpty());
+ }
+
+ @Test
+ void samplesWithoutWatermarkReturnNothingSoDeviceHistoryNeverEntersTheTest()
+ {
+ NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
+ assertTrue(adapter.samplesSince(CHANNEL, null).isEmpty());
+ }
+
+ @Test
+ void emptyWatermarkMeansTheChannelHadNoRowsSoEveryLaterRowBelongsToTheReception()
+ {
+ // 边界时该通道在来源里没有任何行（例如通道尚未开始测试），之后出现的行全部属于本次接收。
+ List<CabinetRealtime> rows = adapter(activeDeviceDatabase())
+ .samplesSince(CHANNEL, new SampleWatermark(null, 0L));
+ assertEquals(2, rows.size());
+ }
+
+ @Test
+ void watermarkIsTheLatestRowOfTheLatestTest()
+ {
+ SampleWatermark watermark = adapter(activeDeviceDatabase()).currentWatermark(CHANNEL);
+ assertEquals(Instant.parse("2026-09-28T01:54:56Z"), watermark.deviceTime());
+ assertEquals(7436L, watermark.sequenceId());
+ // 通道没有测试记录时水位为空，接收路径不把设备历史并入本测试。
+ NewareMysqlAdapter empty = adapter(new FakeVendorDatabase().registeredChannel(DEV_UID, 1, 2));
+ assertNull(empty.currentWatermark(CHANNEL).deviceTime());
+ }
+
+ @Test
+ void rejectsChannelIdentitiesThatBelongToAnotherInstance()
+ {
+ NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
+ ChannelIdentity foreign = new ChannelIdentity("other-source", 21, 36, 1, 2);
+ org.junit.jupiter.api.Assertions.assertThrows(RuntimeException.class, () -> adapter.latestCabinet(foreign));
+ assertTrue(adapter.discoverChannels().stream()
+ .allMatch(item -> ADAPTER_CODE.equals(item.identity().adapterCode())));
+ }
+}
diff --git a/battery-device-neware/src/test/java/com/battery/device/neware/ReadOnlyVendorDataSourceTest.java b/battery-device-neware/src/test/java/com/battery/device/neware/ReadOnlyVendorDataSourceTest.java
new file mode 100644
index 0000000..9141429
--- /dev/null
+++ b/battery-device-neware/src/test/java/com/battery/device/neware/ReadOnlyVendorDataSourceTest.java
@@ -0,0 +1,161 @@
+package com.battery.device.neware;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.Statement;
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.sql.DataSource;
+import org.junit.jupiter.api.Test;
+import com.battery.device.api.model.AdapterCapabilities;
+import com.battery.device.api.model.ChannelIdentity;
+import com.battery.device.api.registry.AdapterInstanceConfig;
+
+/**
+ * 只读防线测试：非查询语句在到达驱动之前被拒绝，且该数据源在其生命周期内从未发出过非查询语句。
+ */
+class ReadOnlyVendorDataSourceTest
+{
+ private static final String ADAPTER_CODE = "neware76";
+ private static final ChannelIdentity CHANNEL = new ChannelIdentity(ADAPTER_CODE, 21, 36, 1, 2);
+
+ private ReadOnlyVendorDataSource dataSourceOver(FakeVendorDatabase database)
+ {
+ return new ReadOnlyVendorDataSource(database.dataSource(), ADAPTER_CODE);
+ }
+
+ @Test
+ void queryStatementsAreForwardedAndRecorded() throws Exception
+ {
+ ReadOnlyVendorDataSource dataSource = new ReadOnlyVendorDataSource(new FakeVendorDatabase().dataSource(),
+ ADAPTER_CODE);
+ try (Connection connection = dataSource.getConnection();
+ PreparedStatement statement = connection.prepareStatement("select 1 from channel"))
+ {
+ statement.executeQuery();
+ }
+ try (Connection connection = dataSource.getConnection();
+ Statement statement = connection.createStatement())
+ {
+ statement.executeQuery("show tables");
+ statement.executeQuery("describe `channel`");
+ }
+ assertTrue(dataSource.audit().rejectedStatements().isEmpty());
+ assertEquals(3, dataSource.audit().forwardedStatements().size());
+ }
+
+ @Test
+ void nonQueryStatementsAreRejectedBeforeTheyReachTheDriver() throws Exception
+ {
+ FakeVendorDatabase database = new FakeVendorDatabase();
+ ReadOnlyVendorDataSource dataSource = dataSourceOver(database);
+ List<String> nonQueries = List.of("update channel set chl_status = 1",
+ "insert into channel(dev_uid) values(210036)", "delete from channel",
+ "drop table data_210036_2026_09_18", "create index idx on channel(chl_status)",
+ "truncate table channel", "set global read_only = 0", "grant select on *.* to x",
+ "alter table channel add column x int", "call some_procedure()");
+ for (String sql : nonQueries)
+ {
+ assertThrows(NonQueryStatementException.class, () -> {
+ try (Connection connection = dataSource.getConnection();
+ PreparedStatement statement = connection.prepareStatement(sql))
+ {
+ statement.execute();
+ }
+ }, "非查询语句必须被拒绝：" + sql);
+ }
+ assertEquals(nonQueries.size(), dataSource.audit().rejectedStatements().size());
+ assertTrue(database.statements().isEmpty(), "被拒绝的语句不得到达驱动");
+ // 语句级执行同样被拦下，并且没有任何非查询语句到达驱动。
+ assertThrows(NonQueryStatementException.class, () -> {
+ try (Connection connection = dataSource.getConnection(); Statement statement = connection.createStatement())
+ {
+ statement.executeUpdate("update channel set chl_status = 1");
+ }
+ });
+ assertThrows(NonQueryStatementException.class,
+ () -> dataSource.getConnection().prepareCall("{call anything()}"));
+ assertTrue(database.statements().isEmpty(), "被拒绝的语句不得到达驱动");
+ assertEquals(nonQueries.size() + 1, dataSource.audit().rejectedStatements().size());
+ }
+
+ /**
+ * 这条断言对应验收要求：整个适配器生命周期内，该数据源没有发出过任何非查询语句。
+ */
+ @Test
+ void adapterLifecycleNeverSendsANonQueryStatement()
+ {
+ FakeVendorDatabase database = new FakeVendorDatabase()
+ .registeredChannel(210036L, 1, 2)
+ .test(new FakeVendorDatabase.TestRow(210036L, 1, 2, 440L, "2026-09-18 09:00:00",
+ "data_210036_2026_09_18", null))
+ .sample(new FakeVendorDatabase.SampleRow("data_210036_2026_09_18", 440L, 1, 2, 7435L, 3, 12, "2",
+ 300000L, 13733, -11889, 0, 0, 50350024, 1103331, "2026-09-28 09:54:56"));
+ ReadOnlyVendorDataSource guarded = dataSourceOver(database);
+ AdapterInstanceConfig config = new AdapterInstanceConfig(ADAPTER_CODE, NewareMysqlAdapter.ADAPTER_TYPE,
+ "26.1.2.3", 3306, 60000, "Asia/Hong_Kong", true, AdapterCapabilities.observeOnly(),
+ Map.of("database", "bts63"));
+ NewareMysqlAdapter adapter = new NewareMysqlAdapter(config, guarded, () -> { },
+ Clock.fixed(Instant.parse("2026-09-28T02:00:00Z"), ZoneOffset.UTC), 0);
+
+ adapter.discoverChannels();
+ adapter.inquire(Set.of(CHANNEL));
+ adapter.latestCabinet(CHANNEL);
+ adapter.currentWatermark(CHANNEL);
+ adapter.samplesSince(CHANNEL,
+ new com.battery.device.api.TestDataObservationAdapter.SampleWatermark(
+ Instant.parse("2026-09-28T01:00:00Z"), 1L));
+
+ assertTrue(guarded.audit().rejectedStatements().isEmpty(),
+ "适配器不应尝试任何非查询语句：" + guarded.audit().rejectedStatements());
+ assertTrue(!database.statements().isEmpty(), "适配器应当实际读取来源");
+ for (String statement : database.statements())
+ {
+ assertTrue(ReadOnlyVendorDataSource.isQuery(statement), "到达驱动的语句必须是查询：" + statement);
+ }
+ }
+
+ @Test
+ void statementClassificationKeepsQueriesAndRejectsEverythingElse()
+ {
+ assertTrue(ReadOnlyVendorDataSource.isQuery("select dev_uid from channel"));
+ assertTrue(ReadOnlyVendorDataSource.isQuery("\n SELECT 1"));
+ assertTrue(ReadOnlyVendorDataSource.isQuery("/* probe */ show grants"));
+ assertTrue(ReadOnlyVendorDataSource.isQuery("-- probe\n describe channel"));
+ assertTrue(ReadOnlyVendorDataSource.isQuery("desc `channel`"));
+ assertTrue(!ReadOnlyVendorDataSource.isQuery("update channel set chl_status = 0"));
+ assertTrue(!ReadOnlyVendorDataSource.isQuery(""));
+ // 以 select 开头但紧跟字母的标识符不是查询语句。
+ assertTrue(!ReadOnlyVendorDataSource.isQuery("selection_helper()"));
+ }
+
+ @Test
+ void rejectedStatementsNeverReachTheDriverEvenWhenTheApplicationKeepsTrying()
+ {
+ // 连续尝试写操作后，来源侧一条语句都没收到，说明拦截发生在驱动之前。
+ FakeVendorDatabase database = new FakeVendorDatabase();
+ ReadOnlyVendorDataSource dataSource = dataSourceOver(database);
+ for (String sql : List.of("update channel set chl_status = 1", "delete from channel",
+ "insert into channel(dev_uid) values(210036)"))
+ {
+ try
+ {
+ dataSource.getConnection().prepareStatement(sql);
+ }
+ catch (Exception ignored)
+ {
+ // 每条都被白名单挡下。
+ }
+ }
+ assertTrue(database.statements().isEmpty());
+ assertEquals(3, dataSource.audit().rejectedStatements().size());
+ }
+}
diff --git a/battery-device-simulator/src/main/java/com/battery/device/simulator/BtsDevelopmentSimulator.java b/battery-device-simulator/src/main/java/com/battery/device/simulator/BtsDevelopmentSimulator.java
index a7728ba..5fa4690 100644
--- a/battery-device-simulator/src/main/java/com/battery/device/simulator/BtsDevelopmentSimulator.java
+++ b/battery-device-simulator/src/main/java/com/battery/device/simulator/BtsDevelopmentSimulator.java
@@ -7,10 +7,14 @@
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLStreamConstants;
import javax.xml.stream.XMLStreamReader;
+import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.api.model.DeviceConnectionStatus;
import com.battery.device.api.model.DeviceIdentity;
@@ -19,6 +23,7 @@
import com.battery.device.api.model.DetectedDeviceMetadata;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.AdapterCapabilities;
+import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.DeviceObservationAdapter;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.simulator.BtsSimulatorStateStore;
@@ -37,8 +42,12 @@
import com.battery.device.simulator.DevelopmentCollectorSimulation;
import com.battery.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
-/** 提供由开发后端托管的临时只读 BTS 模拟端点。 */
-public final class BtsDevelopmentSimulator implements TestDataObservationAdapter
+/**
+ * 提供由开发后端托管的临时只读 BTS 模拟端点。
+ * 同时实现发现与测试数据观测契约：模拟器直接暴露自身状态，因此发现与取数按同一实例解析，
+ * 不再需要单独注入一个扩展观测 Bean。TCP 端点仍保留给协议层测试使用。
+ */
+public final class BtsDevelopmentSimulator implements DeviceAdapter, TestDataObservationAdapter
{
public static final String HOST = "127.0.0.1";
public static final int DEFAULT_PORT = 15020;
@@ -102,6 +111,34 @@
for (ChannelIdentity requested : channels) result.put(requested, observation(requested));
recordRequest("observe", channels.toString(), result.toString());
return java.util.Map.copyOf(result);
+ }
+
+ /** 直接提供模拟器实时快照；不可观测通道不出现在结果里，与 BTS 协议解析后的语义一致。 */
+ @Override
+ public synchronized Map<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
+ {
+ if (channels == null || channels.isEmpty()) return Map.of();
+ Map<ChannelIdentity, RealtimeSnapshot> result = new LinkedHashMap<>();
+ for (ChannelIdentity requested : channels)
+ {
+ if (!canObserve(requested)) continue;
+ CabinetRealtime measurement = cabinetMeasurement(requested);
+ if (measurement == null) continue;
+ DeviceObservation observed = observation(requested);
+ result.put(requested, new RealtimeSnapshot(requested, observed.workStatus(), measurement.cycleNo(),
+ measurement.stepNo(), measurement.stepType(), null, BigDecimal.valueOf(measurement.currentA()),
+ BigDecimal.valueOf(measurement.voltageV()), BigDecimal.valueOf(measurement.capacityAh()),
+ BigDecimal.valueOf(measurement.energyWh()), totalTimeSeconds(requested),
+ measurement.relativeTimeSeconds(), null, measurement.serverReceivedAt(), Map.of(),
+ DeviceConnectionStatus.ONLINE, measurement.deviceTime(), List.of(), null));
+ }
+ recordRequest("inquire", channels.toString(), result.toString());
+ return Map.copyOf(result);
+ }
+
+ private long totalTimeSeconds(ChannelIdentity identity)
+ {
+ return requireDevice(identity.devid()).channels().get(identity.chlid() - 1).configuration().totalTimeSeconds();
}
private DeviceObservation observation(ChannelIdentity requested)
@@ -368,7 +405,7 @@
/** 请求采集器当前配置值；软件不可访问或设备离线时返回空测量。 */
@Override
- public synchronized CollectorRealtime latestCollector(String collectorCode)
+ public synchronized CollectorRealtime latestCollector(String adapterCode, String collectorCode)
{
CollectorState collector = state.collectors().stream().filter(item -> item.deviceCode().equals(collectorCode))
.findFirst().orElse(null);
diff --git a/battery-device-simulator/src/test/java/com/battery/device/simulator/SimulatorObservationAdapterContractTest.java b/battery-device-simulator/src/test/java/com/battery/device/simulator/SimulatorObservationAdapterContractTest.java
index 9407dfd..a35fb53 100644
--- a/battery-device-simulator/src/test/java/com/battery/device/simulator/SimulatorObservationAdapterContractTest.java
+++ b/battery-device-simulator/src/test/java/com/battery/device/simulator/SimulatorObservationAdapterContractTest.java
@@ -16,7 +16,7 @@
{
simulator.updateChannel(1, 1, true, DeviceWorkStatus.PROTECTED);
simulator.updateCollectorChannelStatus(1, "S-01", DeviceWorkStatus.UNOBSERVABLE);
- var collector = simulator.latestCollector("dev-sim-collector-1");
+ var collector = simulator.latestCollector(BtsDevelopmentSimulator.ADAPTER_CODE, "dev-sim-collector-1");
assertTrue(collector.measurements().stream().noneMatch(item -> item.channelCode().equals("S-01")));
firstSequence = collector.measurements().get(0).sequenceId();
simulator.setSoftwareAccessible(false);
@@ -30,7 +30,7 @@
assertEquals(DeviceWorkStatus.PROTECTED, restored.snapshot().devices().get(0).channels().get(0).workStatus());
assertTrue(restored.snapshot().collectors().get(0).latestMeasurements().isEmpty());
restored.setSoftwareAccessible(true);
- assertTrue(restored.latestCollector("dev-sim-collector-1").measurements().get(0).sequenceId() > firstSequence);
+ assertTrue(restored.latestCollector(BtsDevelopmentSimulator.ADAPTER_CODE, "dev-sim-collector-1").measurements().get(0).sequenceId() > firstSequence);
assertEquals(DeviceWorkStatus.UNOBSERVABLE, restored.snapshot().collectors().get(0).channels().get(0).workStatus());
}
}
@@ -63,14 +63,14 @@
simulator.updateChannel(1, 1, true, DeviceWorkStatus.WORKING);
var first = simulator.latestCabinet(channel);
assertNotNull(first);
- assertFalse(simulator.latestCollector("dev-sim-collector-1").measurements().isEmpty());
+ assertFalse(simulator.latestCollector(BtsDevelopmentSimulator.ADAPTER_CODE, "dev-sim-collector-1").measurements().isEmpty());
simulator.setSoftwareAccessible(false);
assertEquals(channel, simulator.discoverChannels().get(0).identity());
assertTrue(simulator.snapshot().devices().get(0).online());
assertEquals(DeviceWorkStatus.UNOBSERVABLE,
simulator.observe(java.util.Set.of(channel)).get(channel).workStatus());
assertNull(simulator.latestCabinet(channel));
- assertTrue(simulator.latestCollector("dev-sim-collector-1").measurements().isEmpty());
+ assertTrue(simulator.latestCollector(BtsDevelopmentSimulator.ADAPTER_CODE, "dev-sim-collector-1").measurements().isEmpty());
simulator.setSoftwareAccessible(true);
var next = simulator.latestCabinet(channel);
assertTrue(next.sequenceId() > first.sequenceId());
diff --git a/pom.xml b/pom.xml
index f64766f..1913403 100644
--- a/pom.xml
+++ b/pom.xml
@@ -181,6 +181,11 @@
</dependency>
<dependency>
<groupId>com.battery</groupId>
+ <artifactId>battery-device-neware</artifactId>
+ <version>${ruoyi.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.battery</groupId>
<artifactId>battery-device-simulator</artifactId>
<version>${ruoyi.version}</version>
</dependency>
@@ -202,6 +207,7 @@
<module>battery-business</module>
<module>battery-device-api</module>
<module>battery-device-bts</module>
+ <module>battery-device-neware</module>
<module>ruoyi-quartz</module>
<module>ruoyi-generator</module>
<module>ruoyi-common</module>
diff --git a/ruoyi-admin/pom.xml b/ruoyi-admin/pom.xml
index 5209293..6973c13 100644
--- a/ruoyi-admin/pom.xml
+++ b/ruoyi-admin/pom.xml
@@ -81,6 +81,12 @@
<artifactId>battery-device-bts</artifactId>
</dependency>
+ <!-- 直连厂商库的只读适配器实现 -->
+ <dependency>
+ <groupId>com.battery</groupId>
+ <artifactId>battery-device-neware</artifactId>
+ </dependency>
+
<!-- 定时任务-->
<dependency>
<groupId>com.ruoyi</groupId>
diff --git a/ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentRealtimeSnapshotScheduler.java b/ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentRealtimeSnapshotScheduler.java
deleted file mode 100644
index b9be79a..0000000
--- a/ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentRealtimeSnapshotScheduler.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package com.battery.business.runtime;
-
-import org.springframework.scheduling.annotation.Scheduled;
-import org.springframework.stereotype.Component;
-import org.springframework.context.annotation.Profile;
-import com.battery.business.service.ITestDataReceptionService;
-
-/** 开发环境触发数据接收，由业务服务按测试接收状态和采样槽位保存成功观测。 */
-@Component
-@Profile("dev-simulator")
-public class DevelopmentRealtimeSnapshotScheduler
-{
- private final ITestDataReceptionService reception;
-
- public DevelopmentRealtimeSnapshotScheduler(ITestDataReceptionService reception) { this.reception = reception; }
-
- @Scheduled(fixedDelay = 100)
- public void capture() { reception.captureReceivedData(); }
-}
diff --git a/ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentSimulatorConfig.java b/ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentSimulatorConfig.java
index ee12cab..53c5461 100644
--- a/ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentSimulatorConfig.java
+++ b/ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentSimulatorConfig.java
@@ -21,10 +21,12 @@
@Bean
@Profile("dev-simulator")
public ApplicationRunner registerDevelopmentSimulatorAdapter(DeviceAdapterRegistry registry,
+ BtsDevelopmentSimulator simulator,
@Value("${battery.dev-simulator.port:15020}") int simulatorPort)
{
// 固定开发装配不进入 t_device_adapter，直接注册运行时实例。
- return arguments -> registry.synchronize(developmentSimulatorRuntimeConfig(simulatorPort));
+ // 注册模拟器对象本身，使发现与取数落在同一实例上，不再单独注入扩展观测 Bean。
+ return arguments -> registry.register(developmentSimulatorRuntimeConfig(simulatorPort), simulator);
}
private AdapterInstanceConfig developmentSimulatorRuntimeConfig(int simulatorPort)
diff --git a/ruoyi-admin/src/main/java/com/battery/business/runtime/DeviceAdapterRuntimeConfig.java b/ruoyi-admin/src/main/java/com/battery/business/runtime/DeviceAdapterRuntimeConfig.java
index 8cdd91f..b819ad6 100644
--- a/ruoyi-admin/src/main/java/com/battery/business/runtime/DeviceAdapterRuntimeConfig.java
+++ b/ruoyi-admin/src/main/java/com/battery/business/runtime/DeviceAdapterRuntimeConfig.java
@@ -3,29 +3,27 @@
import java.util.List;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Profile;
import org.springframework.scheduling.annotation.EnableScheduling;
import com.battery.device.api.NoopDeviceAdapterFactory;
import com.battery.device.bts.BtsDeviceAdapterFactory;
import com.battery.device.api.registry.DeviceAdapterRegistry;
+import com.battery.device.neware.NewareMysqlAdapterFactory;
-/** 装配单进程设备适配器运行时。 */
+/**
+ * 装配单进程设备适配器运行时。
+ *
+ * <p>注册表按存储形态持有全部可用实现；实例按 {@code t_device_adapter} 配置或代码装配注册。
+ * 同一进程内可并存多个来源实例，取数按通道身份里的适配器编码解析到各自实例。
+ * 某个类型是否真的有实例可用只由配置决定，因此这里不再按 profile 收窄可注册的类型。
+ */
@Configuration
@EnableScheduling
public class DeviceAdapterRuntimeConfig
{
@Bean(destroyMethod = "close")
- @Profile("!dev-simulator & !bts")
- public DeviceAdapterRegistry noopDeviceAdapterRegistry()
- {
- return new DeviceAdapterRegistry(List.of(new NoopDeviceAdapterFactory()));
- }
-
- @Bean(destroyMethod = "close")
- @Profile({ "dev-simulator", "bts" })
public DeviceAdapterRegistry deviceAdapterRegistry()
{
- return new DeviceAdapterRegistry(List.of(new BtsDeviceAdapterFactory()));
+ return new DeviceAdapterRegistry(List.of(new NoopDeviceAdapterFactory(), new BtsDeviceAdapterFactory(),
+ new NewareMysqlAdapterFactory()));
}
-
}
diff --git a/ruoyi-admin/src/main/java/com/battery/business/runtime/RedisDeviceDiscoveryCandidateStore.java b/ruoyi-admin/src/main/java/com/battery/business/runtime/RedisDeviceDiscoveryCandidateStore.java
index 4357819..a21b4b2 100644
--- a/ruoyi-admin/src/main/java/com/battery/business/runtime/RedisDeviceDiscoveryCandidateStore.java
+++ b/ruoyi-admin/src/main/java/com/battery/business/runtime/RedisDeviceDiscoveryCandidateStore.java
@@ -129,7 +129,8 @@
object.getString("deviceCode"), object.getIntValue("devtype"), object.getIntValue("devid"),
object.getString("ip"), object.getString("mac"), object.getString("name"),
object.getString("manufacturer"), object.getString("model"), channels,
- requiredInstant(object, "observedAt"), requiredInstant(object, "expiresAt"));
+ requiredInstant(object, "observedAt"), requiredInstant(object, "expiresAt"),
+ optionalInstant(object, "lastDataAt"));
}
private ChannelCandidate channelCandidate(Object value)
@@ -141,7 +142,8 @@
identity.getIntValue("subdevid"), identity.getIntValue("chlid"));
RealtimeSnapshot latestSnapshot = object.getObject("latestSnapshot", RealtimeSnapshot.class);
return new ChannelCandidate(channelIdentity, object.getBooleanValue("online"),
- DeviceWorkStatus.valueOf(object.getString("workStatus")), requiredInstant(object, "observedAt"), latestSnapshot);
+ DeviceWorkStatus.valueOf(object.getString("workStatus")), requiredInstant(object, "observedAt"),
+ latestSnapshot, optionalInstant(object, "lastDataAt"));
}
private CollectorCandidate collectorCandidate(Object value)
@@ -161,6 +163,12 @@
return value;
}
+ /** 最近数据时间只在厂商存储来源上存在，缺失是正常取值。 */
+ private Instant optionalInstant(JSONObject object, String field)
+ {
+ return object.getObject(field, Instant.class);
+ }
+
private void save(StoredState state)
{
redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(state));
diff --git a/ruoyi-admin/src/main/java/com/battery/business/runtime/TestDataReceptionScheduler.java b/ruoyi-admin/src/main/java/com/battery/business/runtime/TestDataReceptionScheduler.java
new file mode 100644
index 0000000..f0766d2
--- /dev/null
+++ b/ruoyi-admin/src/main/java/com/battery/business/runtime/TestDataReceptionScheduler.java
@@ -0,0 +1,22 @@
+package com.battery.business.runtime;
+
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import com.battery.business.service.ITestDataReceptionService;
+
+/**
+ * 按固定周期触发数据接收，由业务服务按测试接收状态和采样来源保存成功观测。
+ *
+ * <p>它不限定 profile：真实来源的接收路径也必须在非开发 profile 下运行。
+ * 开发模拟器专属的调度（模拟器发现轮询、采集器状态同步）仍只在 dev-simulator 下额外启用。
+ */
+@Component
+public class TestDataReceptionScheduler
+{
+ private final ITestDataReceptionService reception;
+
+ public TestDataReceptionScheduler(ITestDataReceptionService reception) { this.reception = reception; }
+
+ @Scheduled(fixedDelay = 1000)
+ public void capture() { reception.captureReceivedData(); }
+}
diff --git a/ruoyi-admin/src/main/resources/db/migration-final/V22__vendor_source_instance_config.sql b/ruoyi-admin/src/main/resources/db/migration-final/V22__vendor_source_instance_config.sql
new file mode 100644
index 0000000..17ef70d
--- /dev/null
+++ b/ruoyi-admin/src/main/resources/db/migration-final/V22__vendor_source_instance_config.sql
@@ -0,0 +1,3 @@
+-- Ticket01：适配器实例的类型按存储形态放宽到直连厂商库，来源形态配置改用 JSON 列承载（ADR-0065）。
+-- 连接口令不进库：source_config 只放 host、port、database、user、时区这类非敏感项。
+ALTER TABLE t_device_adapter ADD COLUMN source_config JSON NULL AFTER capabilities;
diff --git a/ruoyi-admin/src/test/java/com/battery/business/runtime/DeviceAdapterRuntimeConfigTest.java b/ruoyi-admin/src/test/java/com/battery/business/runtime/DeviceAdapterRuntimeConfigTest.java
index 932c945..99f5714 100644
--- a/ruoyi-admin/src/test/java/com/battery/business/runtime/DeviceAdapterRuntimeConfigTest.java
+++ b/ruoyi-admin/src/test/java/com/battery/business/runtime/DeviceAdapterRuntimeConfigTest.java
@@ -8,19 +8,22 @@
import org.mockito.ArgumentCaptor;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.device.api.registry.DeviceAdapterRegistry;
+import com.battery.device.simulator.BtsDevelopmentSimulator;
class DeviceAdapterRuntimeConfigTest
{
@Test
- void registersDevelopmentSimulatorBeforeApplicationReady() throws Exception
+ void registersDevelopmentSimulatorInstanceBeforeApplicationReady() throws Exception
{
DeviceAdapterRegistry registry = mock(DeviceAdapterRegistry.class);
- var runner = new DevelopmentSimulatorConfig().registerDevelopmentSimulatorAdapter(registry, 15021);
+ BtsDevelopmentSimulator simulator = mock(BtsDevelopmentSimulator.class);
+ var runner = new DevelopmentSimulatorConfig().registerDevelopmentSimulatorAdapter(registry, simulator, 15021);
runner.run(new DefaultApplicationArguments(new String[0]));
+ // 注册模拟器对象本身：发现与取数落在同一实例上，不再单独注入扩展观测 Bean。
ArgumentCaptor<AdapterInstanceConfig> adapter = ArgumentCaptor.forClass(AdapterInstanceConfig.class);
- verify(registry).synchronize(adapter.capture());
+ verify(registry).register(adapter.capture(), org.mockito.ArgumentMatchers.same(simulator));
assertEquals("dev-sim", adapter.getValue().adapterCode());
assertEquals("127.0.0.1", adapter.getValue().host());
assertEquals(15021, adapter.getValue().port());
diff --git a/ruoyi-admin/src/test/java/com/battery/business/runtime/VendorSourceInstanceMigrationContractTest.java b/ruoyi-admin/src/test/java/com/battery/business/runtime/VendorSourceInstanceMigrationContractTest.java
new file mode 100644
index 0000000..643bb07
--- /dev/null
+++ b/ruoyi-admin/src/test/java/com/battery/business/runtime/VendorSourceInstanceMigrationContractTest.java
@@ -0,0 +1,97 @@
+package com.battery.business.runtime;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
+import org.junit.jupiter.api.Test;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.init.ScriptUtils;
+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;
+
+/**
+ * 适配器实例按存储形态放宽与来源形态配置列的迁移契约：
+ * 迁移把 {@code source_config} 加在能力列之后，运行库上该列确实存在且可空；
+ * 类型白名单放宽到直连厂商库，未知类型仍然拒绝。
+ */
+class VendorSourceInstanceMigrationContractTest
+{
+ @Test
+ void migrationAddsNullableSourceConfigColumnWithoutTouchingExistingRows() throws Exception
+ {
+ String path = "db/migration-final/V22__vendor_source_instance_config.sql";
+ String migration = resource(path);
+ assertTrue(migration.contains("ALTER TABLE t_device_adapter ADD COLUMN source_config JSON NULL"));
+ assertTrue(migration.contains("AFTER capabilities"));
+ // 迁移只加列：不改既有列、不写入任何数据，更不写入凭据。
+ assertFalse(migration.matches("(?s).*\\b(INSERT|UPDATE|DELETE|DROP)\\b.*"));
+ assertFalse(migration.toLowerCase().contains("password"));
+
+ var source = new UnpooledDataSource("org.h2.Driver",
+ "jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DATABASE_TO_LOWER=TRUE", "sa", "");
+ try (var connection = source.getConnection())
+ {
+ JdbcTemplate jdbc = new JdbcTemplate(source);
+ jdbc.execute("create table t_device_adapter(adapter_id bigint auto_increment primary key,"
+ + "adapter_code varchar(64), capabilities json)");
+ ScriptUtils.executeSqlScript(connection, new ClassPathResource(path));
+
+ // 来源形态配置只承载非敏感项；连接口令由环境变量提供，不入库。
+ jdbc.update("insert into t_device_adapter(adapter_code, capabilities, source_config) values(?,?,?)",
+ "neware76", "{\"control\":false}", "{\"database\":\"bts63\",\"user\":\"visitor\"}");
+ assertEquals(1, jdbc.queryForObject(
+ "select count(*) from t_device_adapter where source_config is not null", Integer.class));
+ // 代码装配实例（例如开发模拟器）保持为空。
+ jdbc.update("insert into t_device_adapter(adapter_code, capabilities) values(?,?)", "dev-sim",
+ "{\"control\":false}");
+ assertEquals(1, jdbc.queryForObject(
+ "select count(*) from t_device_adapter where source_config is null", Integer.class));
+ }
+ }
+
+ @Test
+ void sourceConfigCarriesOnlyNonSensitiveOptionsAndUnknownTypesStayRejected()
+ {
+ AdapterInstanceConfig config = new AdapterInstanceConfig("neware76", "NEWARE_MYSQL", "26.1.2.3", 3306,
+ 60000, "Asia/Hong_Kong", true, AdapterCapabilities.observeOnly(),
+ Map.of("database", "bts63", "user", "visitor", "passwordEnv", "BATTERY_VENDOR_DB_PASSWORD"));
+
+ assertEquals("bts63", config.sourceOption("database"));
+ assertTrue(config.capabilities().discovery());
+ assertFalse(config.capabilities().control());
+
+ try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of()))
+ {
+ // 放宽不等于接受任何类型：未注册的存储形态仍然明确失败。
+ assertThrows(AdapterRegistryException.class,
+ () -> registry.synchronize(new AdapterInstanceConfig("mystery", "SOME_OTHER_VENDOR",
+ "127.0.0.1", 3306, 60000, "Asia/Hong_Kong", true,
+ AdapterCapabilities.observeOnly(), Map.of())));
+ }
+ }
+
+ private static String resource(String path)
+ {
+ try (InputStream stream = VendorSourceInstanceMigrationContractTest.class.getClassLoader()
+ .getResourceAsStream(path))
+ {
+ if (stream == null) throw new IllegalStateException("缺少迁移文件：" + path);
+ return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ catch (java.io.IOException exception)
+ {
+ throw new IllegalStateException("无法读取迁移文件：" + path, exception);
+ }
+ }
+}
diff --git a/ruoyi-admin/src/test/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceServiceTest.java b/ruoyi-admin/src/test/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceServiceTest.java
index dda94d2..a68bc03 100644
--- a/ruoyi-admin/src/test/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceServiceTest.java
+++ b/ruoyi-admin/src/test/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceServiceTest.java
@@ -76,4 +76,40 @@
assertEquals(1, candidateStore.snapshot(java.time.Instant.now()).cabinets().size());
verify(adapter).discoverChannels();
}
+
+ /**
+ * 直连厂商存储的来源：通道登记是长期事实，候选不按时间过期，
+ * 每条候选附带最近一次采样行的设备时间，供操作员判断哪几台真的在产出数据。
+ */
+ @Test
+ void vendorSourceCandidatesCarryLastDataTimeAndDoNotExpireByTime()
+ {
+ InMemoryDeviceDiscoveryCandidateStore candidateStore = new InMemoryDeviceDiscoveryCandidateStore();
+ TDetectionCabinetMapper cabinetMapper = Mockito.mock(TDetectionCabinetMapper.class);
+ TTestMapper testMapper = Mockito.mock(TTestMapper.class);
+ DeviceAdapterRegistry registry = Mockito.mock(DeviceAdapterRegistry.class);
+ DeviceAdapter adapter = Mockito.mock(DeviceAdapter.class);
+ ChannelIdentity identity = new ChannelIdentity("neware76", 21, 36, 1, 2);
+ java.time.Instant lastDataAt = java.time.Instant.parse("2026-09-28T01:54:56Z");
+ when(testMapper.hasReceivingTestForAdapter("neware76")).thenReturn(0);
+ when(adapter.discoverChannels()).thenReturn(List.of(new DetectedChannel(identity, "26.1.2.3", true,
+ com.battery.device.api.model.DetectedDeviceMetadata.empty(), lastDataAt)));
+ when(registry.execute(eq("neware76"), any())).thenAnswer(invocation -> {
+ @SuppressWarnings("unchecked")
+ Function<DeviceAdapter, Object> operation = (Function<DeviceAdapter, Object>) invocation.getArgument(1);
+ return operation.apply(adapter);
+ });
+
+ DeviceDiscoveryPersistenceService service = new DeviceDiscoveryPersistenceService(candidateStore,
+ cabinetMapper, testMapper);
+ service.discoverForReception(registry, new DeviceDiscoveryPersistenceService.DiscoverySource(
+ "neware76", 8L, "新威 7.6 那台", 60000, true));
+
+ var candidate = candidateStore.snapshot(java.time.Instant.now()).cabinets().get(0);
+ assertEquals(lastDataAt, candidate.lastDataAt());
+ assertEquals(lastDataAt, candidate.channels().get(0).lastDataAt());
+ // 一年后仍然是候选：候选只按"已登记且未删除"排除，不按时间过期。
+ assertEquals(1, candidateStore.snapshot(java.time.Instant.now().plusSeconds(365L * 24 * 3600))
+ .cabinets().size());
+ }
}
diff --git a/ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingAdapterWiringContractTest.java b/ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingAdapterWiringContractTest.java
new file mode 100644
index 0000000..3b36194
--- /dev/null
+++ b/ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingAdapterWiringContractTest.java
@@ -0,0 +1,33 @@
+package com.battery.business.service.impl;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.context.annotation.Primary;
+import org.springframework.stereotype.Service;
+import com.battery.device.api.TestDataObservationAdapter;
+
+/**
+ * 取数入口的装配契约：开发模拟器既是 Spring Bean，又实现了测试数据观测契约，
+ * 所以业务侧注入的那个 Bean 必须显式声明为首选，否则上下文里出现两个候选会启动失败。
+ * 这里断言的是"哪个 Bean 会被注入"这个结构性事实，而不是 Spring 内部行为。
+ */
+class RoutingAdapterWiringContractTest
+{
+ @Test
+ void routingAdapterIsThePreferredTestDataObservationBean()
+ {
+ assertTrue(TestDataObservationAdapter.class.isAssignableFrom(RoutingTestDataObservationAdapter.class));
+ assertTrue(RoutingTestDataObservationAdapter.class.isAnnotationPresent(Service.class),
+ "取数入口必须是可注入的 Bean");
+ assertTrue(RoutingTestDataObservationAdapter.class.isAnnotationPresent(Primary.class),
+ "来源实例自己也实现该契约时必须显式首选路由入口，否则注入出现两个候选");
+ }
+
+ @Test
+ void developmentSimulatorAlsoExposesTheContractWhichIsWhyPreferenceIsNeeded()
+ {
+ assertTrue(TestDataObservationAdapter.class
+ .isAssignableFrom(com.battery.device.simulator.BtsDevelopmentSimulator.class));
+ }
+}
diff --git a/ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingTestDataObservationAdapterTest.java b/ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingTestDataObservationAdapterTest.java
new file mode 100644
index 0000000..bf25dfc
--- /dev/null
+++ b/ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingTestDataObservationAdapterTest.java
@@ -0,0 +1,163 @@
+package com.battery.business.service.impl;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import com.battery.device.api.DeviceAdapter;
+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.RealtimeSnapshot;
+import com.battery.device.api.registry.AdapterInstanceConfig;
+import com.battery.device.api.registry.AdapterRegistryException;
+import com.battery.device.api.registry.DeviceAdapterRegistry;
+
+/**
+ * 多来源实例：每个通道的取数请求只送到它所属的实例；实例缺失时明确失败，不回退到其他实例。
+ */
+class RoutingTestDataObservationAdapterTest
+{
+ @Test
+ void sendsEachChannelRequestToTheInstanceNamedByItsAdapterCode()
+ {
+ DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of());
+ try
+ {
+ RoutingTestDataObservationAdapter router = new RoutingTestDataObservationAdapter(registry);
+ RecordingAdapter first = new RecordingAdapter("neware76", 1L);
+ RecordingAdapter second = new RecordingAdapter("neware80", 2L);
+ registry.register(config("neware76"), first);
+ registry.register(config("neware80"), second);
+
+ assertEquals(1L, router.latestCabinet(channel("neware76")).sequenceId());
+ assertEquals(2L, router.latestCabinet(channel("neware80")).sequenceId());
+ assertEquals(List.of("neware76"), first.calls());
+ assertEquals(List.of("neware80"), second.calls());
+ assertEquals(60000L, router.persistenceIntervalMillis(channel("neware76")));
+ assertEquals(70000L, router.persistenceIntervalMillis(channel("neware80")));
+ }
+ finally
+ {
+ registry.close();
+ }
+ }
+
+ @Test
+ void missingInstanceFailsExplicitlyInsteadOfFallingBackToAnotherInstance()
+ {
+ DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of());
+ try
+ {
+ RoutingTestDataObservationAdapter router = new RoutingTestDataObservationAdapter(registry);
+ RecordingAdapter only = new RecordingAdapter("neware76", 1L);
+ registry.register(config("neware76"), only);
+
+ // 实例缺失：明确失败，且没有落到已注册的实例上。
+ assertThrows(AdapterRegistryException.class, () -> router.latestCabinet(channel("neware80")));
+ assertThrows(AdapterRegistryException.class, () -> router.samplesSince(channel("neware80"), null));
+ assertThrows(AdapterRegistryException.class, () -> router.currentWatermark(channel("neware80")));
+ assertTrue(only.calls().isEmpty(), "缺失实例的请求不得落到其他实例");
+ }
+ finally
+ {
+ registry.close();
+ }
+ }
+
+ @Test
+ void aggregatedAdapterDoesNotPretendToDiscoverDevices()
+ {
+ DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of());
+ try
+ {
+ RoutingTestDataObservationAdapter router = new RoutingTestDataObservationAdapter(registry);
+ assertThrows(UnsupportedOperationException.class, router::discoverChannels);
+ assertEquals(RoutingTestDataObservationAdapter.ADAPTER_CODE, router.adapterCode());
+ assertFalse(router.capabilities().control());
+ }
+ finally
+ {
+ registry.close();
+ }
+ }
+
+ private static ChannelIdentity channel(String adapterCode)
+ {
+ return new ChannelIdentity(adapterCode, 21, 36, 1, 2);
+ }
+
+ private static AdapterInstanceConfig config(String adapterCode)
+ {
+ return new AdapterInstanceConfig(adapterCode, "NEWARE_MYSQL", "127.0.0.1", 3306, 60000,
+ "Asia/Hong_Kong", true, AdapterCapabilities.observeOnly());
+ }
+
+ /** 记录被调用次数的实例替身；返回的序号用来区分请求落到了哪一个实例。 */
+ private static final class RecordingAdapter implements DeviceAdapter, TestDataObservationAdapter
+ {
+ private final String code;
+ private final long sequenceId;
+ private final List<String> calls = new ArrayList<>();
+
+ private RecordingAdapter(String code, long sequenceId)
+ {
+ this.code = code;
+ this.sequenceId = sequenceId;
+ }
+
+ private List<String> calls()
+ {
+ return List.copyOf(calls);
+ }
+
+ @Override public String adapterCode() { return code; }
+ @Override public AdapterCapabilities capabilities() { return AdapterCapabilities.observeOnly(); }
+ @Override public List<DetectedChannel> discoverChannels() { return List.of(); }
+ @Override public Map<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
+ {
+ return Map.of();
+ }
+
+ @Override public CabinetRealtime latestCabinet(ChannelIdentity channel)
+ {
+ calls.add(code);
+ return new CabinetRealtime(sequenceId, Instant.EPOCH, 0, 0, 0, 0, 0, 0, null, null, Instant.EPOCH,
+ null, "Asia/Hong_Kong", null);
+ }
+
+ @Override public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
+ {
+ return null;
+ }
+
+ @Override public List<StepRecord> stepRecords(ChannelIdentity channel) { return List.of(); }
+ @Override public List<CycleRecord> cycleRecords(ChannelIdentity channel) { return List.of(); }
+
+ @Override public long persistenceIntervalMillis(ChannelIdentity channel)
+ {
+ calls.add(code);
+ return "neware76".equals(code) ? 60000L : 70000L;
+ }
+
+ @Override public SampleWatermark currentWatermark(ChannelIdentity channel)
+ {
+ calls.add(code);
+ return new SampleWatermark(Instant.EPOCH, 1L);
+ }
+
+ @Override public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
+ {
+ calls.add(code);
+ return List.of();
+ }
+ }
+}
diff --git a/ruoyi-admin/src/test/java/com/battery/business/service/impl/TTestLiveServiceImplTest.java b/ruoyi-admin/src/test/java/com/battery/business/service/impl/TTestLiveServiceImplTest.java
index c9ad4a5..c216676 100644
--- a/ruoyi-admin/src/test/java/com/battery/business/service/impl/TTestLiveServiceImplTest.java
+++ b/ruoyi-admin/src/test/java/com/battery/business/service/impl/TTestLiveServiceImplTest.java
@@ -66,7 +66,7 @@
for (Instant deviceTime : List.of(now.minusSeconds(1), now)) {
var measurement = new TestDataObservationAdapter.CollectorMeasurement("S-01", 1L, deviceTime, deviceTime.toString(),
now, "TEMPERATURE", new BigDecimal("38"), "℃");
- when(adapter.latestCollector("collector-7")).thenReturn(new TestDataObservationAdapter.CollectorRealtime(
+ when(adapter.latestCollector(org.mockito.ArgumentMatchers.nullable(String.class), org.mockito.ArgumentMatchers.eq("collector-7"))).thenReturn(new TestDataObservationAdapter.CollectorRealtime(
"collector-7", true, 1000, List.of(measurement), null));
assertEquals(deviceTime.equals(now) ? 1 : 0, service.current(10L).exceptions().size());
}
@@ -391,7 +391,7 @@
when(simulator.testPlan(any(ChannelIdentity.class))).thenReturn(plan);
if (mapper != null && test.getBatteries().get(0).getCollectorId() != null) {
when(mapper.selectCollectorCode(7L)).thenReturn("collector-7");
- when(simulator.latestCollector("collector-7")).thenReturn(
+ when(simulator.latestCollector(org.mockito.ArgumentMatchers.nullable(String.class), org.mockito.ArgumentMatchers.eq("collector-7"))).thenReturn(
new TestDataObservationAdapter.CollectorRealtime("collector-7", online, 1000, List.of(), null));
}
diff --git a/ruoyi-admin/src/test/java/com/battery/business/service/impl/TestDataReceptionServiceImplTest.java b/ruoyi-admin/src/test/java/com/battery/business/service/impl/TestDataReceptionServiceImplTest.java
index 540bb5b..c6eb367 100644
--- a/ruoyi-admin/src/test/java/com/battery/business/service/impl/TestDataReceptionServiceImplTest.java
+++ b/ruoyi-admin/src/test/java/com/battery/business/service/impl/TestDataReceptionServiceImplTest.java
@@ -59,13 +59,13 @@
Clock clock = mock(Clock.class);
when(clock.instant()).thenReturn(CAPTURED_AT);
var service = new TestDataReceptionServiceImpl(fixture.mapper, fixture.tests, fixture.simulator, clock);
- when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true,
+ when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true,
List.of(collector("S-01", 10L, "temperature", "25.1", "℃"))));
when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);
service.captureReceivedData();
when(clock.instant()).thenReturn(CAPTURED_AT.plusSeconds(5));
- when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true,
+ when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true,
List.of(collector("S-01", 11L, "temperature", "38", "℃"))));
service.captureReceivedData();
@@ -87,7 +87,7 @@
test.getBatteries().get(0).setCollectorChannels(List.of(
mapping("PRIMARY_TEMPERATURE", "S-01"), mapping("SECONDARY_TEMPERATURE_2", "S-03"),
mapping("SECONDARY_TEMPERATURE_3", "S-05")));
- when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true, List.of(
+ when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true, List.of(
collector("S-01", 10L, "temperature", "38", "℃"),
collector("S-03", 10L, "temperature", "25.3", "℃"),
collector("S-05", 10L, "temperature", "25.5", "℃"))));
@@ -125,7 +125,7 @@
{
Fixture fixture = fixture();
when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenReturn(cabinetRecord());
- when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true, List.of(
+ when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true, List.of(
collector("S-01", 10L, "TEMPERATURE", "25.100", "℃"),
collector("S-02", 10L, "POSITIVE_POTENTIAL", "1.620", "V"),
collector("S-03", 10L, "NEGATIVE_POTENTIAL", "1.610", "V"),
@@ -159,7 +159,7 @@
{
Fixture fixture = fixture();
when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenThrow(new IllegalArgumentException("offline"));
- when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true,
+ when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true,
List.of(collector("S-01", 11L, "TEMPERATURE", "26.2", "℃"))));
when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);
@@ -216,7 +216,7 @@
{
Fixture fixture = fixture();
when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenReturn(cabinetRecord());
- when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(false, List.of()));
+ when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(false, List.of()));
when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);
fixture.service().captureReceivedData();
@@ -235,7 +235,7 @@
{
Fixture fixture = fixture();
when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenReturn(cabinetRecord());
- when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(false, List.of()));
+ when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(false, List.of()));
when(fixture.mapper.selectRealtimeRecordContentHash(eq(1L), eq(4L), any(Date.class)))
.thenReturn("different-content");
diff --git a/ruoyi-admin/src/test/java/com/battery/business/service/impl/VendorSampleReceptionServiceTest.java b/ruoyi-admin/src/test/java/com/battery/business/service/impl/VendorSampleReceptionServiceTest.java
new file mode 100644
index 0000000..8f54989
--- /dev/null
+++ b/ruoyi-admin/src/test/java/com/battery/business/service/impl/VendorSampleReceptionServiceTest.java
@@ -0,0 +1,261 @@
+package com.battery.business.service.impl;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import com.battery.device.api.TestDataObservationAdapter;
+import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
+import com.battery.device.api.TestDataObservationAdapter.CollectorRealtime;
+import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
+import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
+import com.battery.device.api.TestDataObservationAdapter.StepRecord;
+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.business.domain.TTest;
+import com.battery.business.domain.TTestBattery;
+import com.battery.business.domain.TTestChannelBinding;
+import com.battery.business.domain.TTestRealtimeRecord;
+import com.battery.business.mapper.TTestMapper;
+import com.battery.business.service.ITTestService;
+
+/**
+ * 逐行接收的归属边界：水位由"开始接收"与"继续接收"这一刻来源里的最新一行决定，
+ * 只取严格晚于水位的数据行；同秒多行都保留；重复同内容幂等；同身份不同内容不覆盖。
+ */
+class VendorSampleReceptionServiceTest
+{
+ private static final Instant START = Instant.parse("2026-09-28T02:00:00Z");
+ private static final ChannelIdentity CHANNEL = new ChannelIdentity("neware76", 21, 36, 1, 2);
+ private static final String IDENTITY = "neware76:21:36:1:2";
+
+ @Test
+ void firstPollReadsTheWatermarkWithoutWritingDeviceHistory()
+ {
+ Fixture fixture = fixture();
+ fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
+
+ fixture.service.captureReceivedData();
+
+ assertEquals(1, fixture.adapter.watermarkReads);
+ assertTrue(fixture.adapter.sampleReads.isEmpty());
+ verify(fixture.mapper, never()).insertRealtimeRecord(any());
+ }
+
+ @Test
+ void everyRowStrictlyAfterTheWatermarkBecomesATestFactIncludingSameSecondRows()
+ {
+ Fixture fixture = fixture();
+ fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
+ fixture.service.captureReceivedData();
+
+ // 同秒两行都落库：槽位留空，来源行身份把它们区分开。
+ fixture.adapter.samples = List.of(sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"),
+ sample(7438L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"));
+ fixture.clock.advance(Duration.ofSeconds(60));
+ fixture.service.captureReceivedData();
+
+ ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
+ verify(fixture.mapper, times(2)).insertRealtimeRecord(rows.capture());
+ assertEquals(List.of("neware:440:7437", "neware:440:7438"),
+ rows.getAllValues().stream().map(TTestRealtimeRecord::getSourceRecordId).toList());
+ assertTrue(rows.getAllValues().stream().allMatch(row -> row.getCaptureSlotUtc() == null));
+ assertEquals(Instant.parse("2026-09-28T01:55:56Z"),
+ rows.getAllValues().get(0).getCabinetDeviceTimeUtc().toInstant());
+ assertEquals("2026-09-28 09:55:56", rows.getAllValues().get(0).getCabinetDeviceTimeRaw());
+ assertEquals(IDENTITY, rows.getAllValues().get(0).getSourceIdentity());
+ }
+
+ @Test
+ void resumingReceptionReadsANewWatermarkSoThePauseGapIsNotBackfilled()
+ {
+ Fixture fixture = fixture();
+ fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
+ fixture.service.captureReceivedData();
+
+ fixture.adapter.samples = List.of(sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"));
+ fixture.clock.advance(Duration.ofSeconds(60));
+ fixture.service.captureReceivedData();
+ assertEquals(List.of(new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L)),
+ fixture.adapter.sampleReads);
+
+ // 暂停后继续：接收边界推进，水位重新读取；暂停期间产生的行不再进本测试。
+ fixture.test.setReceptionBoundaryAtUtc(Date.from(START.plusSeconds(600)));
+ fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T02:05:56Z"), 7500L);
+ fixture.adapter.samples = List.of();
+ fixture.clock.advance(Duration.ofSeconds(60));
+ fixture.service.captureReceivedData();
+
+ assertEquals(2, fixture.adapter.watermarkReads);
+ // 取到新水位后，只追加新水位之后的行；暂停期间的空档不会被补回来。
+ fixture.clock.advance(Duration.ofSeconds(60));
+ fixture.service.captureReceivedData();
+ assertEquals(List.of(new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L),
+ new SampleWatermark(Instant.parse("2026-09-28T02:05:56Z"), 7500L)), fixture.adapter.sampleReads);
+ }
+
+ @Test
+ void repeatedIdenticalRowIsIgnoredAndConflictingContentIsNotOverwritten()
+ {
+ Fixture fixture = fixture();
+ fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
+ fixture.service.captureReceivedData();
+
+ CabinetRealtime sample = sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56");
+ fixture.adapter.samples = List.of(sample);
+ fixture.clock.advance(Duration.ofSeconds(60));
+ fixture.service.captureReceivedData();
+ ArgumentCaptor<TTestRealtimeRecord> first = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
+ verify(fixture.mapper).insertRealtimeRecord(first.capture());
+
+ // 同身份同内容：幂等忽略，不再插入。
+ when(fixture.mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"))
+ .thenReturn(HexFormat.of().formatHex(first.getValue().getContentSha256()));
+ fixture.adapter.watermark = new SampleWatermark(sample.deviceTime(), sample.sequenceId());
+ fixture.clock.advance(Duration.ofSeconds(60));
+ fixture.service.captureReceivedData();
+ verify(fixture.mapper, times(1)).insertRealtimeRecord(any());
+
+ // 同身份不同内容：保留原事实，拒绝覆盖。
+ when(fixture.mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"))
+ .thenReturn("0000000000000000000000000000000000000000000000000000000000000000");
+ fixture.clock.advance(Duration.ofSeconds(60));
+ fixture.service.captureReceivedData();
+ verify(fixture.mapper, times(1)).insertRealtimeRecord(any());
+ }
+
+ @Test
+ void sourceWithoutWatermarkKeepsThePollingSlotPath()
+ {
+ Fixture fixture = fixture();
+ // 不提供水位的来源（例如开发模拟器）继续按轮询槽位保存柜端事实。
+ fixture.adapter.watermark = null;
+ fixture.adapter.latest = sample(9L, "2026-09-28T02:00:30Z", "2026-09-28 10:00:30");
+
+ fixture.service.captureReceivedData();
+
+ assertEquals(1, fixture.adapter.watermarkReads);
+ assertTrue(fixture.adapter.sampleReads.isEmpty());
+ ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
+ verify(fixture.mapper).insertRealtimeRecord(rows.capture());
+ assertTrue(rows.getValue().getCaptureSlotUtc() != null);
+ }
+
+ private static CabinetRealtime sample(long sequenceId, String deviceTime, String raw)
+ {
+ return new CabinetRealtime(sequenceId, Instant.parse(deviceTime), -1.1889, 1.3733, 1.398611, 0.030648,
+ 3, 12, null, "2", Instant.parse("2026-09-28T02:01:00Z"), raw, "Asia/Hong_Kong", 300L,
+ "neware:440:" + sequenceId);
+ }
+
+ private static Fixture fixture()
+ {
+ TTestMapper mapper = mock(TTestMapper.class);
+ ITTestService tests = mock(ITTestService.class);
+ AdaptiveAdapter adapter = new AdaptiveAdapter();
+ TTest test = receivingTest();
+ when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
+ when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
+ when(mapper.selectCollectorCode(9L)).thenReturn("collector-9");
+ when(tests.selectTestById(1L)).thenReturn(test);
+ TestClock clock = new TestClock(START);
+ return new Fixture(mapper, tests, adapter, test, clock,
+ new TestDataReceptionServiceImpl(mapper, tests, adapter, clock));
+ }
+
+ private static TTest receivingTest()
+ {
+ TTestChannelBinding binding = new TTestChannelBinding();
+ binding.setBindingId(2L);
+ binding.setChannelId(3L);
+ binding.setAdapterCode("neware76");
+ binding.setAdapterTimezoneId("Asia/Hong_Kong");
+ binding.setDevtype(21);
+ binding.setDevid(36);
+ binding.setSubdevid(1);
+ binding.setChlid(2);
+ TTestBattery battery = new TTestBattery();
+ battery.setTestBatteryId(4L);
+ battery.setCollectorId(9L);
+ battery.setBinding(binding);
+ battery.setCollectorChannels(List.of());
+ TTest test = new TTest();
+ test.setTestId(1L);
+ test.setTestStatus("RUNNING");
+ test.setReceptionStatus("RECEIVING");
+ test.setStartedAtUtc(Date.from(START));
+ test.setReceptionBoundaryAtUtc(Date.from(START));
+ test.setBatteries(List.of(battery));
+ return test;
+ }
+
+ private record Fixture(TTestMapper mapper, ITTestService tests, AdaptiveAdapter adapter, TTest test,
+ TestClock clock, TestDataReceptionServiceImpl service) { }
+
+ /** 只暴露水位与逐行读取的测试替身，并记录被问到的水位与逐行读取次数。 */
+ private static final class AdaptiveAdapter implements TestDataObservationAdapter
+ {
+ private SampleWatermark watermark;
+ private List<CabinetRealtime> samples = new ArrayList<>();
+ private CabinetRealtime latest;
+ private int watermarkReads;
+ private final List<SampleWatermark> sampleReads = new ArrayList<>();
+
+ @Override public String adapterCode() { return "neware76"; }
+ @Override public AdapterCapabilities capabilities() { return AdapterCapabilities.observeOnly(); }
+ @Override public List<DetectedChannel> discoverChannels() { return List.of(); }
+ @Override public Map<ChannelIdentity, DeviceObservation> observe(Set<ChannelIdentity> channels)
+ { return Map.of(); }
+ @Override public CabinetRealtime latestCabinet(ChannelIdentity channel) { return latest; }
+ @Override public CollectorRealtime latestCollector(String adapterCode, String collectorCode) { return null; }
+ @Override public List<StepRecord> stepRecords(ChannelIdentity channel) { return List.of(); }
+ @Override public List<CycleRecord> cycleRecords(ChannelIdentity channel) { return List.of(); }
+ @Override public long persistenceIntervalMillis(ChannelIdentity channel) { return 60000L; }
+
+ @Override public SampleWatermark currentWatermark(ChannelIdentity channel)
+ {
+ watermarkReads++;
+ return watermark;
+ }
+
+ @Override public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark since)
+ {
+ sampleReads.add(since);
+ return List.copyOf(samples);
+ }
+ }
+
+ /** 可推进的测试时钟：接收节奏与厂商写库节奏都按它推进。 */
+ private static final class TestClock extends Clock
+ {
+ private Instant now;
+
+ private TestClock(Instant now) { this.now = now; }
+
+ private void advance(Duration duration) { now = now.plus(duration); }
+
+ @Override public ZoneId getZone() { return ZoneOffset.UTC; }
+ @Override public Clock withZone(ZoneId zone) { return this; }
+ @Override public Instant instant() { return now; }
+ }
+}
diff --git a/ruoyi-admin/src/test/java/com/battery/business/web/VendorSamplePersistenceDatabaseTest.java b/ruoyi-admin/src/test/java/com/battery/business/web/VendorSamplePersistenceDatabaseTest.java
new file mode 100644
index 0000000..d32e525
--- /dev/null
+++ b/ruoyi-admin/src/test/java/com/battery/business/web/VendorSamplePersistenceDatabaseTest.java
@@ -0,0 +1,156 @@
+package com.battery.business.web;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.time.Instant;
+import java.util.Date;
+import java.util.HexFormat;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mybatis.spring.SqlSessionFactoryBean;
+import org.mybatis.spring.SqlSessionTemplate;
+import org.springframework.jdbc.core.JdbcTemplate;
+import com.battery.business.domain.TTestRealtimeRecord;
+import com.battery.business.mapper.TTestMapper;
+
+/**
+ * 逐行接收的落库语义（生产 Mapper + H2 内存库）：
+ * 设备数据行不占系统轮询槽位，所以同一绝对秒内的多行可以各自成立；
+ * 来源行身份（厂商测试编号 + 协议序号）承担幂等与冲突判定。
+ */
+public class VendorSamplePersistenceDatabaseTest
+{
+ private static final String IDENTITY = "neware76:21:36:1:2";
+
+ private java.sql.Connection keepAlive;
+ private JdbcTemplate jdbc;
+ private TTestMapper mapper;
+
+ @BeforeEach void setUp() throws Exception
+ {
+ var source = new UnpooledDataSource("org.h2.Driver",
+ "jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DATABASE_TO_LOWER=TRUE", "sa", "");
+ keepAlive = source.getConnection();
+ jdbc = new JdbcTemplate(source);
+ // H2 无 MySQL 的 HEX()；测试中补一个等价别名，生产 SQL 仍在 MySQL 上验证。
+ jdbc.execute("create alias HEX for 'com.battery.business.web.VendorSamplePersistenceDatabaseTest.hexOf'");
+ jdbc.execute("create table t_test_realtime_record(record_id bigint auto_increment primary key,"
+ + "test_id bigint, test_battery_id bigint, binding_id bigint, adapter_code varchar(64),"
+ + "source_identity varchar(256) not null, source_record_id varchar(128),"
+ + "collector_id bigint, collector_channel_code varchar(100), collector_sequence_id bigint,"
+ + "collector_device_time_utc timestamp(3), collector_device_time_raw varchar(64),"
+ + "collector_adapter_timezone_id varchar(64),"
+ + "cabinet_devtype int, cabinet_devid int, cabinet_subdevid int, cabinet_chlid int,"
+ + "cabinet_sequence_id bigint, cabinet_device_time_utc timestamp(3), cabinet_device_time_raw varchar(64),"
+ + "cabinet_adapter_timezone_id varchar(64), cycle_no int, step_no int, step_type varchar(64),"
+ + "step_name varchar(191), voltage_v decimal(30,12), current_a decimal(30,12),"
+ + "capacity_ah decimal(30,12), energy_wh decimal(30,12), temperature decimal(30,12),"
+ + "positive_potential decimal(30,12), negative_potential decimal(30,12),"
+ + "extension_measurements_json varchar(2000), capture_slot_utc timestamp(3) null, "
+ + "server_received_at_utc timestamp(3) not null, content_sha256 binary(32) not null,"
+ // 轮询槽位唯一键允许多个空值：逐行接收因此可以把同一秒的多行都保存下来。
+ + "unique(test_id, test_battery_id, capture_slot_utc))");
+
+ SqlSessionFactoryBean factory = new SqlSessionFactoryBean();
+ factory.setDataSource(source);
+ factory.setTypeAliasesPackage("com.battery.business.domain");
+ factory.setMapperLocations(new org.springframework.core.io.ClassPathResource("mapper/battery/TTestMapper.xml"));
+ mapper = new SqlSessionTemplate(factory.getObject()).getMapper(TTestMapper.class);
+ }
+
+ @AfterEach void closeDatabase() throws Exception { if (keepAlive != null) keepAlive.close(); }
+
+ @Test void sameSecondRowsAreKeptSeparatelyAndRepeatIsRecognisedBySourceRowIdentity()
+ {
+ TTestRealtimeRecord first = row(7437L, "2026-09-28 09:55:56", "2026-09-28T01:55:56Z");
+ assertEquals(1, mapper.insertRealtimeRecord(first));
+
+ // 同一绝对秒的第二行：槽位为空，唯一键不冲突，两行都成立。
+ TTestRealtimeRecord second = row(7438L, "2026-09-28 09:55:56", "2026-09-28T01:55:56Z");
+ assertEquals(1, mapper.insertRealtimeRecord(second));
+
+ assertEquals(2, jdbc.queryForObject("select count(*) from t_test_realtime_record", Integer.class));
+ // 幂等与冲突按来源行身份判定：命中同身份即视为重复，由调用方比较内容摘要。
+ assertEquals(hex(first.getContentSha256()),
+ mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"));
+ assertNull(mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7439"));
+ assertTrue(!hex(first.getContentSha256()).equals(hex(second.getContentSha256())));
+ }
+
+ @Test void deviceRowKeepsVendorRawTimeServerReceiveTimeAndNoPollingSlot()
+ {
+ mapper.insertRealtimeRecord(row(7437L, "2026-09-28 09:55:56", "2026-09-28T01:55:56Z"));
+
+ Map<String, Object> stored = jdbc.queryForMap("select source_record_id, source_identity,"
+ + " cabinet_device_time_raw, cabinet_device_time_utc, server_received_at_utc, capture_slot_utc,"
+ + " cabinet_adapter_timezone_id, extension_measurements_json, voltage_v"
+ + " from t_test_realtime_record");
+ assertEquals("neware:440:7437", stored.get("source_record_id"));
+ assertEquals(IDENTITY, stored.get("source_identity"));
+ // 设备时间按适配器时区解释后以 UTC 保存，同时保留原始时间文本与服务器接收时间。
+ assertEquals("2026-09-28 09:55:56", stored.get("cabinet_device_time_raw"));
+ assertEquals("Asia/Hong_Kong", stored.get("cabinet_adapter_timezone_id"));
+ assertEquals(Instant.parse("2026-09-28T01:55:56Z"),
+ ((java.sql.Timestamp) stored.get("cabinet_device_time_utc")).toInstant());
+ assertEquals(Instant.parse("2026-09-28T02:01:00Z"),
+ ((java.sql.Timestamp) stored.get("server_received_at_utc")).toInstant());
+ assertNull(stored.get("capture_slot_utc"));
+ // H2 的 JSON 列会把 JSON 文本再引一层，这里只断言内容确实保存下来。
+ assertTrue(String.valueOf(stored.get("extension_measurements_json")).contains("relativeTimeSeconds"));
+ assertEquals(0, new BigDecimal("1.3733").compareTo((BigDecimal) stored.get("voltage_v")));
+ }
+
+ /** 供 H2 别名使用的 HEX 等价实现；H2 没有 MySQL 的 HEX()。 */
+ public static String hexOf(byte[] value)
+ {
+ return value == null ? null : HexFormat.of().formatHex(value);
+ }
+
+ private TTestRealtimeRecord row(long sequenceId, String raw, String deviceTimeUtc)
+ {
+ TTestRealtimeRecord row = new TTestRealtimeRecord();
+ row.setTestId(1L); row.setTestBatteryId(4L); row.setBindingId(2L); row.setAdapterCode("neware76");
+ row.setSourceIdentity(IDENTITY); row.setSourceRecordId("neware:440:" + sequenceId);
+ row.setCabinetDevtype(21); row.setCabinetDevid(36); row.setCabinetSubdevid(1); row.setCabinetChlid(2);
+ row.setCabinetSequenceId(sequenceId);
+ row.setCabinetDeviceTimeUtc(Date.from(Instant.parse(deviceTimeUtc)));
+ row.setCabinetDeviceTimeRaw(raw);
+ row.setCabinetAdapterTimezoneId("Asia/Hong_Kong");
+ row.setCaptureSlotUtc(null);
+ row.setServerReceivedAtUtc(Date.from(Instant.parse("2026-09-28T02:01:00Z")));
+ row.setCycleNo(3); row.setStepNo(12); row.setStepType("2");
+ row.setVoltageV(new BigDecimal("1.3733")); row.setCurrentA(new BigDecimal("-1.1889"));
+ row.setCapacityAh(new BigDecimal("1.398612")); row.setEnergyWh(new BigDecimal("0.030648"));
+ row.setExtensionMeasurementsJson("{\"relativeTimeSeconds\":300}");
+ // 内容摘要把协议序号算进去，使同秒不同行的内容彼此不同。
+ row.setContentSha256(sha256("neware:440:" + sequenceId));
+ return row;
+ }
+
+ private static String hex(byte[] value)
+ {
+ return HexFormat.of().formatHex(value);
+ }
+
+ private static byte[] sha256(String value)
+ {
+ try
+ {
+ return MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8));
+ }
+ catch (Exception exception)
+ {
+ throw new IllegalStateException(exception);
+ }
+ }
+
+}
--
Gitblit v1.7.1