From 1de35281a699b8d2ca90777f63c7e032c3c9dc3d Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期一, 12 十月 2026 13:34:34 +0800
Subject: [PATCH] 对接第 6 台新威上位机（7.6 客户端，走库路线），并修两处对接中暴露的问题
---
battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsAdapter.java | 67 +++++++++++++++++++++-
scripts/register-neware-6th-source.sql | 32 ++++++++++
battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsPipeTransport.java | 24 +++++++
battery-device-bts/src/test/java/com/battery/device/bts/NewareBtsAdapterContractTest.java | 45 ++++++++++++--
4 files changed, 155 insertions(+), 13 deletions(-)
diff --git a/battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsAdapter.java b/battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsAdapter.java
index 5ec2f15..d85470e 100644
--- a/battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsAdapter.java
+++ b/battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsAdapter.java
@@ -62,6 +62,11 @@
private final AdapterCapabilities capabilities;
/** 每通道最近一次工步层读取；见 {@link #stepLayer}。 */
private final Map<String, StepLayer> stepLayers = new ConcurrentHashMap<>();
+ /**
+ * 该客户端有没有数据文件命令（{@code inquiredf} / {@code download} / {@code downloadStepLayer}）。
+ * 未知为 null；探测一次后固定下来——见 {@link #supportsDataFileCommands}。
+ */
+ private volatile Boolean dataFileCommandsSupported;
public NewareBtsAdapter(String adapterCode, BtsTransport transport, BtsXmlCodec codec)
{
@@ -191,6 +196,12 @@
}
catch (BtsProtocolException unsupported)
{
+ // 客户端不提供 inquiredf（实测 7.6.0.1475）：本来源不按设备数据行接收，逐点回落到轮询槽位。
+ return null;
+ }
+ catch (BtsTransportException unanswered)
+ {
+ // 探测说支持、发出去却不作答（已登记不再试）：同样回落到轮询槽位，不把来源不可达说成"不按行接收"。
return null;
}
if (state.count() < 1) return new SampleWatermark(null, 0L);
@@ -256,9 +267,19 @@
private StepLayer fetchStepLayer(ChannelIdentity channel)
{
long fetchedAtNanos = System.nanoTime();
+ // 客户端没有这条命令时一条都不发：见 supportsDataFileCommands。
+ if (!supportsDataFileCommands(channel)) return new StepLayer(List.of(), null, fetchedAtNanos);
Instant receivedAt = Instant.now();
- List<StepRecord> layers = codec.parseStepLayer(
- transport.exchange(codec.downloadStepLayerRequest(channel)), receivedAt);
+ List<StepRecord> layers;
+ try
+ {
+ layers = codec.parseStepLayer(transport.exchange(codec.downloadStepLayerRequest(channel)), receivedAt);
+ }
+ catch (BtsTransportException unanswered)
+ {
+ rememberUnanswered(unanswered);
+ return new StepLayer(List.of(), null, fetchedAtNanos);
+ }
if (layers.isEmpty()) return new StepLayer(List.of(), null, fetchedAtNanos);
// 通道仍在动（运行、暂停、保护中）时最后一步没结束：它和它所在的循环都先按住。
if (!isStillRunning(channel)) return new StepLayer(layers, null, fetchedAtNanos);
@@ -343,7 +364,18 @@
private BtsXmlCodec.DataFileState dataFileState(ChannelIdentity channel)
{
- return codec.parseDataFileState(transport.exchange(codec.inquiredfRequest(channel)));
+ if (!supportsDataFileCommands(channel))
+ throw new BtsProtocolException(BtsProtocolException.Reason.MISSING_FIELD,
+ "该客户端不提供 inquiredf");
+ try
+ {
+ return codec.parseDataFileState(transport.exchange(codec.inquiredfRequest(channel)));
+ }
+ catch (BtsTransportException unanswered)
+ {
+ rememberUnanswered(unanswered);
+ throw unanswered;
+ }
}
private List<CabinetRealtime> download(ChannelIdentity channel, long testId, long startPosition, int count)
@@ -352,6 +384,35 @@
Instant.now());
}
+ /**
+ * 这台客户端有没有数据文件命令（{@code inquiredf}/{@code download}/{@code downloadStepLayer}）。
+ *
+ * <p>为什么必须先问一句再发：**对不支持这些命令的客户端发出去，代价不只是超时**。2026-10-12 实测
+ * （客户端 7.6.0.1475）：发一条 {@code inquiredf} 之后，该接口的 {@code inquire} 连续 8 分钟无应答
+ * （{@code getdevinfo} 仍正常）——那等于干扰了那台上位机的对外接口。所以不支持的客户端
+ * **一条都不发**，逐点回落到轮询槽位、工步层为空。
+ *
+ * <p>探测依据是 {@code inquire} 应答的字段：7.6 的应答不带 {@code cycle_id}，8.0 的带；
+ * 实测三个构建（7.6.0.1475 / 8.0.0.484 / 8.0.0.530）里"有没有 {@code cycle_id}"与
+ * "有没有这三条数据文件命令"同向。探测本身读不到应答时**抛出**（不吞）：那说明来源不可达，
+ * 按契约要抛异常，不能被当成"该来源不按数据行接收"。
+ */
+ private boolean supportsDataFileCommands(ChannelIdentity channel)
+ {
+ Boolean known = dataFileCommandsSupported;
+ if (known != null) return known;
+ RealtimeSnapshot snapshot = inquire(Set.of(channel)).get(channel);
+ boolean supported = snapshot != null && snapshot.cycleNo() != null;
+ dataFileCommandsSupported = supported;
+ return supported;
+ }
+
+ /** 探测说支持、真发出去却不作答：记下来不再反复试——反复发一条无应答的命令会持续干扰那台接口。 */
+ private void rememberUnanswered(BtsTransportException failure)
+ {
+ if (failure.reason() == BtsTransportException.Reason.TIMEOUT) dataFileCommandsSupported = false;
+ }
+
/** 行的序号比水位大就是要的行；序号反而更小说明换了测试，此时按设备时间判定。 */
private static boolean isAfter(SampleWatermark watermark, CabinetRealtime row)
{
diff --git a/battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsPipeTransport.java b/battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsPipeTransport.java
index 1d1bc0e..afcc6ab 100644
--- a/battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsPipeTransport.java
+++ b/battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsPipeTransport.java
@@ -145,14 +145,14 @@
}
catch (TimeoutException ex)
{
- closeInFlight();
+ interruptRead();
pending.cancel(true);
throw new BtsTransportException(BtsTransportException.Reason.TIMEOUT,
"BTS 管道应答超时（" + responseTimeoutMs + " ms）：" + pipePath, ex);
}
catch (InterruptedException ex)
{
- closeInFlight();
+ interruptRead();
pending.cancel(true);
Thread.currentThread().interrupt();
throw new BtsTransportException(BtsTransportException.Reason.IO_FAILURE,
@@ -347,6 +347,26 @@
try { pipe.close(); } catch (IOException ignored) { }
}
+ /**
+ * 在独立线程里关掉在途句柄，用来打断阻塞中的读。
+ *
+ * <p>为什么不自己关：实测关闭一条"服务端不作答的连接"**本身会阻塞**（关闭要等 SMB 层回收），
+ * 那样超时就白设了——调用线程照样被钉死。而这个实例是串行执行的（BTS 一条连接只服务一条请求，
+ * 注册表也按实例串行），一条不响应的命令足以把该实例之后的所有请求堵住。
+ * 关闭线程是守护线程且不等它结束：最坏情况是每次超时留一个挂着的关闭线程，
+ * 而超时本身应当罕见——真出现"某客户端不支持某命令"，它的表现正是每次都无应答。
+ */
+ private void interruptRead()
+ {
+ RandomAccessFile pipe = inFlight;
+ if (pipe == null) return;
+ Thread closer = new Thread(() -> {
+ try { pipe.close(); } catch (IOException ignored) { }
+ }, "neware-bts-pipe-close");
+ closer.setDaemon(true);
+ closer.start();
+ }
+
@Override
public void close()
{
diff --git a/battery-device-bts/src/test/java/com/battery/device/bts/NewareBtsAdapterContractTest.java b/battery-device-bts/src/test/java/com/battery/device/bts/NewareBtsAdapterContractTest.java
index d75dfc1..f848477 100644
--- a/battery-device-bts/src/test/java/com/battery/device/bts/NewareBtsAdapterContractTest.java
+++ b/battery-device-bts/src/test/java/com/battery/device/bts/NewareBtsAdapterContractTest.java
@@ -86,7 +86,7 @@
@Test
void watermarkIsTheLastPersistedRowAndSamplesAreStrictlyAfterIt()
{
- ScriptedPipe pipe = new ScriptedPipe().rows(1, 5);
+ ScriptedPipe pipe = new ScriptedPipe().rows(1, 5).on("inquire", CAPABLE_CLIENT);
try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
{
SampleWatermark watermark = adapter.currentWatermark(CHANNEL);
@@ -94,7 +94,8 @@
// 水位 = 已落盘行数那一行本身；行数就是最后一行的 seqid（实测一致）。
assertEquals(5L, watermark.sequenceId());
assertEquals(rowTime(5), watermark.deviceTime());
- assertEquals(List.of("inquiredf", "download:5:1"), pipe.commands());
+ // 先探测一次客户端能力（inquire 带 cycle_id → 支持数据文件命令），再取水位。
+ assertEquals(List.of("inquire", "inquiredf", "download:5:1"), pipe.commands());
pipe.reset();
var rows = adapter.samplesSince(CHANNEL, new SampleWatermark(rowTime(3), 3L));
@@ -120,10 +121,10 @@
}
@Test
- void treatsAClientWithoutInquiredfAsNotRowBasedInsteadOfFailing()
+ void treatsAnInquiredfReplyWithoutARecordAsNotRowBased()
{
- // 实测 7.6.0.1475 不支持 inquiredf（应答里没有 chl 记录）。这不是失败，是该构建的真实能力。
- ScriptedPipe pipe = new ScriptedPipe().on("inquiredf",
+ // 应答里没有 chl 记录（命令看似支持却没有该通道的数据）→ 不按数据行接收，回落轮询槽位。
+ ScriptedPipe pipe = new ScriptedPipe().on("inquire", CAPABLE_CLIENT).on("inquiredf",
"<bts><cmd>inquiredf_resp</cmd><list count=\"0\" /></bts>");
try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
{
@@ -132,9 +133,32 @@
}
@Test
+ void doesNotSendDataFileCommandsToAClientWithoutCycleId()
+ {
+ // 实测 7.6.0.1475：`inquire` 应答不带 cycle_id，而它没有 inquiredf/download/downloadStepLayer。
+ // 向它发 inquiredf 的代价不只是超时——实测会让该接口的 `inquire` 连续 8 分钟无应答
+ // （getdevinfo 仍正常），等于干扰了那台上位机的对外接口。所以一条都不发。
+ ScriptedPipe pipe = new ScriptedPipe().on("inquire", "<bts><cmd>inquire_resp</cmd><list count=\"1\">"
+ + "<inquire dev=\"21-103-1-1-0\" step_id=\"5\" step_type=\"rest\" workstatus=\"working\""
+ + " voltage=\"1.2531\" current=\"0.0000\" auxtemp=\"0\" auxvol=\"0\" /></list></bts>");
+ try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
+ {
+ assertNull(adapter.currentWatermark(CHANNEL));
+ assertTrue(adapter.stepRecords(CHANNEL).isEmpty());
+ assertTrue(adapter.cycleRecords(CHANNEL).isEmpty());
+
+ assertFalse(pipe.commands().contains("inquiredf"));
+ assertFalse(pipe.commands().contains("download"));
+ assertFalse(pipe.commands().contains("downloadsteplayer"));
+ // 探测只做一次，之后一直按"不支持"处理。
+ assertEquals(List.of("inquire"), pipe.commands());
+ }
+ }
+
+ @Test
void aChannelWithNoTestHasAZeroWatermarkAndNoSamples()
{
- ScriptedPipe pipe = new ScriptedPipe().on("inquiredf",
+ ScriptedPipe pipe = new ScriptedPipe().on("inquire", CAPABLE_CLIENT).on("inquiredf",
"<bts><cmd>inquiredf_resp</cmd><list count=\"1\"><chl devtype=\"25\" devid=\"81\" subdevid=\"1\""
+ " chlid=\"2\" testid=\"316\" count=\"0\">true</chl></list></bts>");
try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
@@ -243,17 +267,22 @@
@Test
void aChannelWithNoTestInTheBufferHasNoStepsOrCycles()
{
- ScriptedPipe pipe = new ScriptedPipe()
+ ScriptedPipe pipe = new ScriptedPipe().on("inquire", CAPABLE_CLIENT)
.on("downloadStepLayer", "<bts><cmd>downloadStepLayer_resp</cmd><list count=\"0\" /></bts>");
try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
{
assertTrue(adapter.stepRecords(CHANNEL).isEmpty());
assertTrue(adapter.cycleRecords(CHANNEL).isEmpty());
// 缓冲里没有测试就直接是空，不必再去问工作状态。
- assertEquals(List.of("downloadsteplayer"), pipe.commands());
+ assertEquals(List.of("inquire", "downloadsteplayer"), pipe.commands());
}
}
+ /** 一个"支持数据文件命令"的客户端应答（8.0 系：`inquire` 带 `cycle_id`）。 */
+ private static final String CAPABLE_CLIENT = "<bts><cmd>inquire_resp</cmd><list count=\"1\">"
+ + "<inquire dev=\"25-81-1-2-0\" cycle_id=\"1\" step_id=\"5\" step_type=\"rest\""
+ + " workstatus=\"working\" voltage=\"1.25\" /></list></bts>";
+
@Test
void aFailedStateReadSkipsTheRoundInsteadOfHandingOutTheRunningStep()
{
diff --git a/scripts/register-neware-6th-source.sql b/scripts/register-neware-6th-source.sql
new file mode 100644
index 0000000..6d1b250
--- /dev/null
+++ b/scripts/register-neware-6th-source.sql
@@ -0,0 +1,32 @@
+-- 登记第 6 台新威上位机（26.8.215.130）：客户端 7.6 + 服务端 7.6 形态，走直连厂商库。
+-- 在你本地的**业务库**里执行一次即可。
+--
+-- 为什么是 NEWARE_MYSQL 而不是 NEWARE_BTS：
+-- 1. 客户端是 7.6（实测：`inquire` 字段集无 cycle_id/log_code、有 auxtemp/auxvol；
+-- `getdevinfo_resp` 的 `<server>` 无 port 属性）→ `inquiredf`/`download`/`downloadStepLayer`
+-- 在该构建上无应答，接口侧拿不到逐点与工步；
+-- 2. 服务端是 7.6 形态（库内 482 张 `data_*` 逐点表）→ 逐点与工步都能从库读。
+-- 即与 A 那台（`neware76`）同类。
+--
+-- 更硬的一条理由（2026-10-12 实测）：对 7.6 客户端发 `inquiredf` 会让该接口的 `inquire`
+-- 长时间无应答（实测 8 分钟未恢复，`getdevinfo` 仍正常）——**别把这台登记成 NEWARE_BTS**。
+--
+-- source_config 只放非敏感项：口令在后端 config/vendor-credentials.properties 的
+-- BATTERY_VENDOR_DB_PASSWORD（现场取得的服务端只读账号 visitor）。
+
+INSERT INTO t_device_adapter
+ (adapter_code, adapter_name, adapter_type, host, port, poll_interval_ms, timezone_id, read_only,
+ capabilities, connection_status, status, del_flag, source_config, create_by, remark)
+SELECT 'neware-b', '新威 B（26.8.215.130）', 'NEWARE_MYSQL', '26.8.215.130', 3306,
+ r.poll_interval_ms, r.timezone_id, r.read_only, r.capabilities, 'OFFLINE', '0', 0,
+ '{"user":"visitor","database":"bts63","passwordEnv":"BATTERY_VENDOR_DB_PASSWORD"}',
+ 'system', '7.6 客户端 + 7.6 服务端形态；6 台设备（dev_deleted=0）、48 通道'
+FROM (
+ SELECT poll_interval_ms, timezone_id, read_only, capabilities
+ FROM t_device_adapter
+ WHERE adapter_code = 'neware76' AND del_flag = 0
+ ) r;
+
+-- 核对：应看到 1 行 neware-b（NEWARE_MYSQL）
+SELECT adapter_id, adapter_code, adapter_name, adapter_type, host, port, timezone_id, status, source_config
+FROM t_device_adapter WHERE adapter_code = 'neware-b';
--
Gitblit v1.7.1