java/BatteryDetection.git

package com.battery.device.bts;
 
import java.io.ByteArrayOutputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.CodingErrorAction;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Locale;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.ReentrantLock;
 
/**
 * 新威 BTS 对外接口的**远程命名管道**传输（ADR-0068）。
 *
 * <p>现场实测的三条关键事实决定了这个类的形状：
 *
 * <ol>
 *   <li><b>接口是命名管道，不是 TCP。</b>协议文档的 TCP 小节在现场没有对应物；
 *       可达的是服务端命名管道 {@code \\.\pipe\NewareBtsAPI}。跨机访问它要走 SMB，
 *       所以这里用 {@code \\<主机>\pipe\NewareBtsAPI} 这个 Win32 路径，
 *       并在打开管道前先建一条 SMB 会话。</li>
 *   <li><b>一条连接只服务一条请求。</b>实测在同一条连接上发第二条命令会永久挂住
 *       （不报错、不返回）。所以每次请求都新建连接、收到应答即关闭，不做长连接复用。</li>
 *   <li><b>报文结束符是纯 {@code \n\n}。</b>文档里的 {@code #\r\n} 与 {@code #\r\n} 变体实测均无效
 *       （后者零响应）。这里按实测取 {@code \n\n}。</li>
 * </ol>
 *
 * <p><b>权限</b>：调用方需管理员身份，且 SMB 会话用的是那台机器的本地管理员账号——
 * 口令从本地凭据文件取（{@link com.battery.device.api.credentials.LocalCredentials}），不进数据库、不进仓库。
 * 会话由本类用 {@code net.exe use} 建立：Java 无法给命名管道传凭据，
 * 只有先有一条已认证的会话，路径才打得开。会话按"主机 + 账号 + 口令"缓存，进程内每条只建一次。
 *
 * <p><b>只读边界</b>：出站请求先过 {@link NewareBtsReadOnlyCommands} 的白名单，
 * 非只读命令在写进管道之前就被拒绝。
 *
 * <p><b>读超时</b>：{@code RandomAccessFile} 没有读超时（管道挂住就是永久阻塞），
 * 因此读取放在单线程执行器里、由调用方带超时等待；超时从另一个线程关掉在途句柄把读打断，
 * 使"挂住"变成一次可分类的失败，而不是把调用线程钉死。
 */
public final class NewareBtsPipeTransport implements BtsTransport
{
    /** 服务端命名管道的名字；跨机时路径为 {@code \\<主机>\pipe\<名字>}。 */
    private static final String PIPE_NAME = "NewareBtsAPI";
    /** 实测的报文结束符（协议文档的 {@code #\r\n} 在本接口上无效）。 */
    private static final byte[] FRAME_END = "\n\n".getBytes(StandardCharsets.UTF_8);
    private static final int DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024 * 1024;
    private static final int MAX_SESSION_TIMEOUT_MS = 30_000;
    private static final int MAX_RESPONSE_TIMEOUT_MS = 120_000;
    private static final int READ_BUFFER_BYTES = 64 * 1024;
    /** 已建立的 SMB 会话；键为「主机 + 账号 + 口令」的摘要，进程内复用。 */
    private static final Set<String> ESTABLISHED_SESSIONS = ConcurrentHashMap.newKeySet();
 
    private final String host;
    private final String user;
    private final String password;
    private final String pipePath;
    private final String sessionKey;
    private final int sessionTimeoutMs;
    private final int responseTimeoutMs;
    private final int maxResponseBytes;
    private final ReentrantLock requestLock = new ReentrantLock(true);
    /** 读取放在这里跑，使调用方能带超时等待；线程是守护线程，不拖住进程退出。 */
    private final ExecutorService reader = Executors.newSingleThreadExecutor(runnable -> {
        Thread thread = new Thread(runnable, "neware-bts-pipe-reader");
        thread.setDaemon(true);
        return thread;
    });
    /** 当前在途的管道句柄：读超时时由调用线程关掉它，把阻塞中的读打断。 */
    private volatile RandomAccessFile inFlight;
 
    public NewareBtsPipeTransport(String host, String user, String password,
            Duration sessionTimeout, Duration responseTimeout)
    {
        this(host, user, password, sessionTimeout, responseTimeout, DEFAULT_MAX_RESPONSE_BYTES);
    }
 
    NewareBtsPipeTransport(String host, String user, String password,
            Duration sessionTimeout, Duration responseTimeout, int maxResponseBytes)
    {
        this.host = requireText(host, "host");
        this.user = requireText(user, "user");
        this.password = Objects.requireNonNull(password, "password");
        this.sessionTimeoutMs = timeout(sessionTimeout, MAX_SESSION_TIMEOUT_MS, "session");
        this.responseTimeoutMs = timeout(responseTimeout, MAX_RESPONSE_TIMEOUT_MS, "response");
        this.maxResponseBytes = maxResponseBytes;
        if (maxResponseBytes < 1) throw new IllegalArgumentException("maxResponseBytes must be positive");
        this.pipePath = "\\\\" + this.host + "\\pipe\\" + PIPE_NAME;
        this.sessionKey = this.host + "|" + this.user + "|" + Integer.toHexString(this.password.hashCode());
    }
 
    /**
     * 发出一条只读请求并返回原始应答。
     *
     * <p>前置条件：请求只含只读白名单内的命令。并发调用由锁串行化（本接口一条连接只服务一条请求）。
     * SMB 会话首次使用时建立；管道打开或读取失败时重建会话并重试一次，仍失败则归类抛出。
     *
     * @param request 完整请求报文（不含结束符）
     * @return 应答原文（已去掉 {@code \n\n}）
     * @throws BtsTransportException 请求越权，或会话、管道、读取失败
     */
    @Override
    public String exchange(String request)
    {
        NewareBtsReadOnlyCommands.requireAllowed(request);
        requestLock.lock();
        try
        {
            ensureSession(false);
            try
            {
                return perform(request);
            }
            catch (BtsTransportException failure)
            {
                // 会话可能已被对端或系统回收（长时间空闲、重启上位机）：重建一次再试，只试一次。
                if (!isSessionRecoverable(failure.reason())) throw failure;
                ensureSession(true);
                return perform(request);
            }
        }
        finally
        {
            requestLock.unlock();
        }
    }
 
    private String perform(String request)
    {
        Future<String> pending = reader.submit(() -> openAndExchange(request));
        try
        {
            return pending.get(responseTimeoutMs, TimeUnit.MILLISECONDS);
        }
        catch (TimeoutException ex)
        {
            closeInFlight();
            pending.cancel(true);
            throw new BtsTransportException(BtsTransportException.Reason.TIMEOUT,
                    "BTS 管道应答超时（" + responseTimeoutMs + " ms）：" + pipePath, ex);
        }
        catch (InterruptedException ex)
        {
            closeInFlight();
            pending.cancel(true);
            Thread.currentThread().interrupt();
            throw new BtsTransportException(BtsTransportException.Reason.IO_FAILURE,
                    "BTS 管道读取被中断：" + pipePath, ex);
        }
        catch (ExecutionException ex)
        {
            Throwable cause = ex.getCause();
            if (cause instanceof BtsTransportException failure) throw failure;
            throw new BtsTransportException(BtsTransportException.Reason.IO_FAILURE,
                    "BTS 管道交换失败：" + pipePath, cause);
        }
    }
 
    /** 每次请求新建一条连接：实测同一连接上的第二条命令不会返回。 */
    private String openAndExchange(String request) throws IOException
    {
        RandomAccessFile pipe;
        try
        {
            pipe = new RandomAccessFile(pipePath, "rw");
        }
        catch (FileNotFoundException ex)
        {
            throw new BtsTransportException(BtsTransportException.Reason.CONNECTION_FAILED,
                    "无法打开新威 BTS 管道（接口未启用？未激活？无管理员凭据？）：" + pipePath, ex);
        }
        inFlight = pipe;
        try
        {
            pipe.write((request + new String(FRAME_END, StandardCharsets.UTF_8)).getBytes(StandardCharsets.UTF_8));
            return decode(readFrame(pipe, maxResponseBytes));
        }
        catch (java.io.EOFException ex)
        {
            throw new BtsTransportException(BtsTransportException.Reason.DISCONNECTED,
                    "新威 BTS 管道在应答结束前关闭：" + pipePath, ex);
        }
        catch (FrameTooLargeException ex)
        {
            throw new BtsTransportException(BtsTransportException.Reason.RESPONSE_TOO_LARGE,
                    "新威 BTS 应答超过 " + maxResponseBytes + " 字节：" + pipePath, ex);
        }
        catch (InvalidUtf8Exception ex)
        {
            throw new BtsTransportException(BtsTransportException.Reason.INVALID_UTF8,
                    "新威 BTS 应答不是合法 UTF-8：" + pipePath, ex);
        }
        finally
        {
            inFlight = null;
            try { pipe.close(); } catch (IOException ignored) { }
        }
    }
 
    /**
     * 按块读取并扫描 {@code \n\n}，返回去掉结束符的应答字节。逐字节读在 SMB 上是一次读一次往返，
     * 而 {@code download} 一页最多上千行，所以必须成块读。
     */
    static byte[] readFrame(RandomAccessFile pipe, int maxPayloadBytes) throws IOException
    {
        byte[] chunk = new byte[READ_BUFFER_BYTES];
        ByteArrayOutputStream received = new ByteArrayOutputStream();
        int matched = 0;
        while (true)
        {
            int read = pipe.read(chunk);
            if (read < 0) throw new java.io.EOFException("新威 BTS 管道在应答结束前关闭");
            for (int i = 0; i < read; i++)
            {
                byte value = chunk[i];
                if (value == FRAME_END[matched])
                {
                    matched++;
                    if (matched == FRAME_END.length) return received.toByteArray();
                    continue;
                }
                if (matched > 0)
                {
                    // 已匹配到的结束符前缀是正文的一部分：原样补回，再按当前字节重判。
                    for (int prefix = 0; prefix < matched; prefix++) append(received, FRAME_END[prefix], maxPayloadBytes);
                    matched = 0;
                }
                if (value == FRAME_END[0])
                {
                    matched = 1;
                    continue;
                }
                append(received, value, maxPayloadBytes);
            }
        }
    }
 
    private static void append(ByteArrayOutputStream buffer, byte value, int maxPayloadBytes) throws FrameTooLargeException
    {
        if (buffer.size() >= maxPayloadBytes) throw new FrameTooLargeException();
        buffer.write(value);
    }
 
    private static String decode(byte[] bytes) throws InvalidUtf8Exception
    {
        try
        {
            return StandardCharsets.UTF_8.newDecoder()
                    .onMalformedInput(CodingErrorAction.REPORT)
                    .onUnmappableCharacter(CodingErrorAction.REPORT)
                    .decode(ByteBuffer.wrap(bytes)).toString();
        }
        catch (CharacterCodingException ex)
        {
            throw new InvalidUtf8Exception(ex);
        }
    }
 
    /**
     * 建立 SMB 会话。Java 无法向命名管道传凭据，所以只能先让系统用 {@code net.exe use} 认证一次，
     * 之后管道路径才打得开；这是"把人工那步 net use 收进后端"的落点。
     *
     * @param force true 表示先撤掉旧会话再建（用于重试路径）
     */
    private void ensureSession(boolean force)
    {
        if (force)
        {
            runNet("use", "\\\\" + host + "\\ipc$", "/delete", "/y");
            ESTABLISHED_SESSIONS.remove(sessionKey);
        }
        if (!force && ESTABLISHED_SESSIONS.contains(sessionKey)) return;
        String output = runNet("use", "\\\\" + host + "\\ipc$", "/user:" + host + "\\" + user, password);
        if (output == null) throw new BtsTransportException(BtsTransportException.Reason.CONNECTION_FAILED,
                "无法建立到 " + host + " 的 SMB 会话", null);
        ESTABLISHED_SESSIONS.add(sessionKey);
    }
 
    /**
     * 执行一条 net.exe 命令。
     *
     * <p>返回值：成功返回输出；失败时若输出表明"该主机的会话已存在"（错误 1219/多重连接）
     * 也按成功处理——会话已经在了，管道照常可开。其余失败返回空值。
     * 建立与撤销都失败不抛出，交给后续管道打开去报错，避免凭据问题被当成会话问题。
     */
    private String runNet(String... arguments)
    {
        String[] command = new String[arguments.length + 1];
        command[0] = netExecutable();
        System.arraycopy(arguments, 0, command, 1, arguments.length);
        try
        {
            Process process = new ProcessBuilder(command).redirectErrorStream(true).start();
            String output = new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8);
            if (!process.waitFor(sessionTimeoutMs, TimeUnit.MILLISECONDS))
            {
                process.destroyForcibly();
                return null;
            }
            if (process.exitValue() == 0) return output;
            String normalized = output.toLowerCase(Locale.ROOT);
            if (normalized.contains("1219") || normalized.contains("多重连接") || normalized.contains("multiple connections"))
                return output;
            return null;
        }
        catch (IOException ex)
        {
            return null;
        }
        catch (InterruptedException ex)
        {
            Thread.currentThread().interrupt();
            return null;
        }
    }
 
    private static String netExecutable()
    {
        String systemRoot = System.getenv("SystemRoot");
        return (systemRoot == null || systemRoot.isBlank() ? "C:\\Windows" : systemRoot) + "\\System32\\net.exe";
    }
 
    /** 会话或连接类失败值得重建会话再试一次；协议与内容类失败重试没有意义。 */
    private static boolean isSessionRecoverable(BtsTransportException.Reason reason)
    {
        return reason == BtsTransportException.Reason.CONNECTION_FAILED
                || reason == BtsTransportException.Reason.DISCONNECTED
                || reason == BtsTransportException.Reason.IO_FAILURE;
    }
 
    /** 关掉在途句柄：这是唯一能打断阻塞中 ReadFile 的手段。 */
    private void closeInFlight()
    {
        RandomAccessFile pipe = inFlight;
        if (pipe == null) return;
        try { pipe.close(); } catch (IOException ignored) { }
    }
 
    @Override
    public void close()
    {
        requestLock.lock();
        try
        {
            closeInFlight();
        }
        finally
        {
            requestLock.unlock();
        }
        reader.shutdownNow();
    }
 
    private static String requireText(String value, String field)
    {
        String normalized = Objects.requireNonNull(value, field).trim();
        if (normalized.isEmpty()) throw new IllegalArgumentException(field + " must not be blank");
        return normalized;
    }
 
    private static int timeout(Duration value, int maximumMs, String field)
    {
        int millis = Math.toIntExact(Objects.requireNonNull(value, field).toMillis());
        if (millis < 1) throw new IllegalArgumentException(field + " timeout must be positive");
        if (millis > maximumMs) throw new IllegalArgumentException(field + " timeout must not exceed " + maximumMs + " ms");
        return millis;
    }
 
    private static final class FrameTooLargeException extends IOException
    {
        private static final long serialVersionUID = 1L;
    }
 
    private static final class InvalidUtf8Exception extends IOException
    {
        private static final long serialVersionUID = 1L;
 
        private InvalidUtf8Exception(Throwable cause)
        {
            super(cause);
        }
    }
}