对接第 6 台新威上位机（7.6 客户端，走库路线），并修两处对接中暴露的问题

yeg

2 天以前 1de35281a699b8d2ca90777f63c7e032c3c9dc3d

对接第 6 台新威上位机（7.6 客户端，走库路线），并修两处对接中暴露的问题

第 6 台（26.8.215.130）：客户端 7.6（inquire 无 cycle_id、getdevinfo 无 port 属性），
服务端 7.6 形态（482 张 data_* 逐点表、visitor@% 可远程只读连）。与 A 那台同类，
因此走 NEWARE_MYSQL，登记为 neware-b（scripts/register-neware-6th-source.sql）。
用生产库适配器实测通过：48 通道、实时值、水位、增量逐点、16 工步、3 循环（效率 91.6%/92.3%）。

对接中暴露并修掉两处：

1. 超时后关闭在途句柄会钉死调用线程（NewareBtsPipeTransport）。
   实测关闭一条"服务端不作答的连接"本身会阻塞，于是超时白设——调用线程照旧被钉死，
   而这个实例是串行执行的，一条不响应的命令足以把该实例之后的所有请求堵住。
   改为在独立守护线程里关句柄、不等它结束。实测：inquiredf 超时 8 秒后调用线程正常返回。

2. 对不支持数据文件命令的客户端发 inquiredf，代价不只是超时（NewareBtsAdapter）。
   实测（7.6.0.1475）：发一条 inquiredf 之后，该接口的 inquire 连续 8 分钟无应答
   （getdevinfo 仍正常），而那台正在跑测试。厂商软件自身不受影响（走另一条内部管道，
   库侧 data_* 持续写入），但这等于干扰了它对外的接口，必须防住。
   现在先探测一次：inquire 应答没有 cycle_id 的客户端**一条数据文件命令都不发**
   （逐点回落轮询槽位、工步层为空）；另加兜底：探测说支持却真发出去无应答时不再重复发。
   探测读不到应答时抛出，以免把"来源不可达"说成"不按行接收"。
   用生产代码在该台验证：实际只发出 [inquire] 一条探测。

battery-device-bts 72 项、后端全量 250 项通过，BUILD SUCCESS。

 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)
    {

 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()
    {

 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()
    {

 scripts/register-neware-6th-source.sql

New file
@@ -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';