gateway/forecast-south/src/main/java/cn/oleauto/south/ApduSlicer.java

@@ -0,0 +1,62 @@
+package cn.oleauto.south;
+
+import java.io.ByteArrayOutputStream;
+import java.util.Arrays;
+import java.util.function.Consumer;
+
+/**
+ * IEC 60870-5-104 APDU 切帧器。
+ *
+ * <p>帧格式：起始字节 0x68，1 字节长度 L（有效范围 4..253），随后 L 字节体，每帧总长 = 2 + L。
+ * 本类有状态，支持跨 chunk 拼接：内部 carry 缓冲携带未消费字节，
+ * {@link #feed(byte[], int, int, Consumer)} 喂入任意 chunk 后，切出的完整 APDU 经 emit 回调输出。
+ *
+ * <p>重新同步规则：扫描到非 0x68 字节或长度非法时，逐字节前推直至找到 0x68。
+ * 正常传输下 pos 总落在帧边界（每次 emit 后直接跳到下一帧），不会误判体内 0x68。
+ */
+public class ApduSlicer {
+
+    private final ByteArrayOutputStream carry = new ByteArrayOutputStream();
+
+    /**
+     * 喂入一段字节，切出的完整 APDU 经 emit 回调输出。
+     *
+     * @param b    字节数组
+     * @param off  起始偏移
+     * @param len  长度
+     * @param emit 完整帧回调（可能被调用 0 次或多次）
+     */
+    public void feed(byte[] b, int off, int len, Consumer<byte[]> emit) {
+        if (len <= 0) {
+            return;
+        }
+        carry.write(b, off, len);
+        byte[] buf = carry.toByteArray();
+        int end = buf.length;
+        int pos = 0;
+        while (pos < end) {
+            if (buf[pos] != 0x68) {
+                pos++;
+                continue;
+            }
+            if (pos + 1 >= end) {
+                break; // 长度字节未到，保留
+            }
+            int l = buf[pos + 1] & 0xFF;
+            if (l < 4 || l > 253) {
+                pos++; // 非法长度，跳过当前字节重新同步
+                continue;
+            }
+            int frameLen = 2 + l;
+            if (pos + frameLen > end) {
+                break; // 帧不完整，保留
+            }
+            emit.accept(Arrays.copyOfRange(buf, pos, pos + frameLen));
+            pos += frameLen;
+        }
+        carry.reset();
+        if (pos < end) {
+            carry.write(buf, pos, end - pos);
+        }
+    }
+}

gateway/forecast-south/src/main/java/cn/oleauto/south/MonitoringInputStream.java

@@ -0,0 +1,75 @@
+package cn.oleauto.south;
+
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.function.Consumer;
+
+/**
+ * 监听输入流：透传原始字节的同时，将 chunk 喂入 {@link ApduSlicer}，
+ * 切出的完整 APDU 经 onFrame 回调输出。读到 EOF 或抛 IOException 时触发一次 onClosed。
+ *
+ * <p>典型链路：BufferedInputStream → 本流 → 原始 socket 输入流。
+ * BufferedInputStream 调用本流 {@link #read(byte[], int, int)} 填充缓冲时完成切帧。
+ */
+public class MonitoringInputStream extends FilterInputStream {
+
+    private final ApduSlicer slicer;
+    private final Consumer<byte[]> onFrame;
+    private final Runnable onClosed;
+    private boolean closedNotified = false;
+
+    public MonitoringInputStream(InputStream in, Consumer<byte[]> onFrame, Runnable onClosed) {
+        super(in);
+        this.onFrame = onFrame;
+        this.onClosed = onClosed;
+        this.slicer = new ApduSlicer();
+    }
+
+    @Override
+    public int read() throws IOException {
+        try {
+            int b = super.read();
+            if (b == -1) {
+                notifyClosed();
+                return -1;
+            }
+            oneByte[0] = (byte) b;
+            slicer.feed(oneByte, 0, 1, onFrame);
+            return b;
+        } catch (IOException e) {
+            notifyClosed();
+            throw e;
+        }
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+        try {
+            int n = super.read(b, off, len);
+            if (n == -1) {
+                notifyClosed();
+                return -1;
+            }
+            if (n > 0) {
+                slicer.feed(b, off, n, onFrame);
+            }
+            return n;
+        } catch (IOException e) {
+            notifyClosed();
+            throw e;
+        }
+    }
+
+    private final byte[] oneByte = new byte[1];
+
+    private void notifyClosed() {
+        if (closedNotified) {
+            return;
+        }
+        closedNotified = true;
+        if (onClosed != null) {
+            onClosed.run();
+        }
+    }
+}

gateway/forecast-south/src/main/java/cn/oleauto/south/MonitoringOutputStream.java

@@ -0,0 +1,45 @@
+package cn.oleauto.south;
+
+import java.io.FilterOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.function.Consumer;
+
+/**
+ * 监听输出流：透传原始字节的同时，将写入的 chunk 喂入 {@link ApduSlicer}，
+ * 切出的完整 APDU 经 onFrame 回调输出。
+ *
+ * <p>典型链路：BufferedOutputStream → 本流 → 原始 socket 输出流。
+ * BufferedOutputStream 刷新缓冲时调用本流 {@link #write(byte[], int, int)}，
+ * 一次刷新可能包含多个 APDU，由 slicer 逐帧输出；也可能在一次刷新中只含半帧，
+ * 由 slicer 的 carry 缓冲跨次拼接。
+ *
+ * <p>注意：直接调用 {@code out.write(b, off, len)} 而非 {@code super.write(...)}，
+ * 以避免 {@link FilterOutputStream} 默认逐字节实现导致重复喂入 slicer。
+ */
+public class MonitoringOutputStream extends FilterOutputStream {
+
+    private final ApduSlicer slicer;
+    private final Consumer<byte[]> onFrame;
+
+    public MonitoringOutputStream(OutputStream out, Consumer<byte[]> onFrame) {
+        super(out);
+        this.onFrame = onFrame;
+        this.slicer = new ApduSlicer();
+    }
+
+    @Override
+    public void write(int b) throws IOException {
+        oneByte[0] = (byte) b;
+        slicer.feed(oneByte, 0, 1, onFrame);
+        out.write(b);
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len) throws IOException {
+        slicer.feed(b, off, len, onFrame);
+        out.write(b, off, len);
+    }
+
+    private final byte[] oneByte = new byte[1];
+}

gateway/forecast-south/src/main/java/cn/oleauto/south/MonitoringSocket.java

@@ -0,0 +1,69 @@
+package cn.oleauto.south;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.util.function.Consumer;
+
+/**
+ * 监听 Socket：继承 {@link Socket}，仅重写 {@link #getInputStream()} / {@link #getOutputStream()}，
+ * 用 {@link MonitoringInputStream} / {@link MonitoringOutputStream} 包装底层流，实现对收发字节的非侵入式监听。
+ *
+ * <p>{@link cn.oleauto.iec60870.ClientConnectionBuilder#build()} 通过 {@link MonitoringSocketFactory#createSocket()}
+ * 创建本类实例后，正常调用 {@code setSoTimeout / bind / connect} 连接本 Socket，
+ * 因此继承自 {@link Socket} 的 {@code getInetAddress / getPort / getRemoteSocketAddress / isConnected / close} 等方法
+ * 均返回本 Socket 的真实连接状态，无需额外转发。
+ *
+ * <p>{@link cn.oleauto.iec60870.Connection} 构造时调用 {@code socket.getInputStream()/getOutputStream()}
+ * 包装为 Buffered* 流——本类的 getInputStream/getOutputStream 返回的监听流位于 Buffered 流与原始 socket 流之间，
+ * 既能捕获真实收发字节，又不影响 Connection 原有读写逻辑。
+ */
+public class MonitoringSocket extends Socket {
+
+    private final Consumer<byte[]> onReceivedFrame;
+    private final Consumer<byte[]> onSentFrame;
+    private final Runnable onClosed;
+
+    private volatile MonitoringInputStream monitoredIn;
+    private volatile MonitoringOutputStream monitoredOut;
+
+    public MonitoringSocket(Consumer<byte[]> onReceivedFrame,
+                            Consumer<byte[]> onSentFrame,
+                            Runnable onClosed) {
+        super();
+        this.onReceivedFrame = onReceivedFrame;
+        this.onSentFrame = onSentFrame;
+        this.onClosed = onClosed;
+    }
+
+    @Override
+    public InputStream getInputStream() throws IOException {
+        MonitoringInputStream local = monitoredIn;
+        if (local == null) {
+            synchronized (this) {
+                local = monitoredIn;
+                if (local == null) {
+                    local = new MonitoringInputStream(super.getInputStream(), onReceivedFrame, onClosed);
+                    monitoredIn = local;
+                }
+            }
+        }
+        return local;
+    }
+
+    @Override
+    public OutputStream getOutputStream() throws IOException {
+        MonitoringOutputStream local = monitoredOut;
+        if (local == null) {
+            synchronized (this) {
+                local = monitoredOut;
+                if (local == null) {
+                    local = new MonitoringOutputStream(super.getOutputStream(), onSentFrame);
+                    monitoredOut = local;
+                }
+            }
+        }
+        return local;
+    }
+}

gateway/forecast-south/src/main/java/cn/oleauto/south/MonitoringSocketFactory.java

@@ -0,0 +1,71 @@
+package cn.oleauto.south;
+
+import javax.net.SocketFactory;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.function.Consumer;
+
+/**
+ * 监听 Socket 工厂：供 {@link cn.oleauto.iec60870.ClientConnectionBuilder#setSocketFactory} 注入。
+ *
+ * <p>{@link ClientConnectionBuilder#build()} 调用 {@link #createSocket()} 获取未连接的 {@link MonitoringSocket}，
+ * 随后自行执行 {@code setSoTimeout / connect}。其余带参重载为完备性实现，均创建 MonitoringSocket 后连接。
+ *
+ * <p>构造参数为三个回调：onReceivedFrame（收到完整 APDU）、onSentFrame（发出完整 APDU）、
+ * onClosed（输入流 EOF 或 IOException）。典型用法见 {@link SouthTransmit#connect()}。
+ */
+public class MonitoringSocketFactory extends SocketFactory {
+
+    private final Consumer<byte[]> onReceivedFrame;
+    private final Consumer<byte[]> onSentFrame;
+    private final Runnable onClosed;
+
+    public MonitoringSocketFactory(Consumer<byte[]> onReceivedFrame,
+                                   Consumer<byte[]> onSentFrame,
+                                   Runnable onClosed) {
+        this.onReceivedFrame = onReceivedFrame;
+        this.onSentFrame = onSentFrame;
+        this.onClosed = onClosed;
+    }
+
+    @Override
+    public Socket createSocket() throws IOException {
+        return new MonitoringSocket(onReceivedFrame, onSentFrame, onClosed);
+    }
+
+    @Override
+    public Socket createSocket(String host, int port) throws IOException {
+        Socket s = createSocket();
+        s.connect(new InetSocketAddress(host, port));
+        return s;
+    }
+
+    @Override
+    public Socket createSocket(String host, int port, InetAddress localHost, int localPort) throws IOException {
+        Socket s = createSocket();
+        if (localHost != null) {
+            s.bind(new InetSocketAddress(localHost, localPort));
+        }
+        s.connect(new InetSocketAddress(host, port));
+        return s;
+    }
+
+    @Override
+    public Socket createSocket(InetAddress host, int port) throws IOException {
+        Socket s = createSocket();
+        s.connect(new InetSocketAddress(host, port));
+        return s;
+    }
+
+    @Override
+    public Socket createSocket(InetAddress address, int port, InetAddress localAddress, int localPort) throws IOException {
+        Socket s = createSocket();
+        if (localAddress != null) {
+            s.bind(new InetSocketAddress(localAddress, localPort));
+        }
+        s.connect(new InetSocketAddress(address, port));
+        return s;
+    }
+}

gateway/forecast-south/src/main/java/cn/oleauto/south/SouthTransmit.java

@@ -0,0 +1,186 @@
+package cn.oleauto.south;
+
+import cn.oleauto.iec60870.ClientConnectionBuilder;
+import cn.oleauto.iec60870.Connection;
+import cn.oleauto.protocol.common.IChannel;
+import cn.oleauto.protocol.common.transmit.AbstractTransmit;
+import cn.oleauto.protocol.common.transmit.TransmitState;
+import cn.oleauto.south.transfer.EFileTransfer;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.StringReader;
+import java.net.InetAddress;
+import java.util.Properties;
+
+/**
+ * 南网 IEC 60870-5-104 传输包装器，{@link cn.oleauto.modbus.rtu.MyRtuMaster} 的对等类。
+ *
+ * <p>包装 iec60870 {@link Connection} 与 {@link EFileTransfer}，实现 {@link cn.oleauto.protocol.common.ITransmit}，
+ * 负责连接建立/断开与状态上报。文件传输类 ASDU 由 {@link EFileTransfer} 自行处理（其 delegate 留空）；
+ * 收发报文通过注入 {@link MonitoringSocketFactory} 在 socket 字节层切帧，
+ * 经 {@link #notifyReceived(byte[])} / {@link #notifySent(byte[])} 触发
+ * {@link cn.oleauto.protocol.common.protocol.AbstractProtocol#onTransmitReceived} /
+ * {@link cn.oleauto.protocol.common.protocol.AbstractProtocol#onTransmitSent}，从而进入 MQ 报文上报。
+ *
+ * <p>配置项（Properties 字符串，由 {@link #start(String)} 解析）：
+ * <pre>host=127.0.0.1
+ * port=2404
+ * commonAddress=1
+ * informationObjectAddress=0</pre>
+ *
+ * <p>{@link #send(byte[])} / {@link #receive()} 不支持：IEC 104 使用 ASDU/文件传输，而非原始字节收发，
+ * 数据交互请通过 {@link #getFileTransfer()} 调用 {@link EFileTransfer#sendFile}。
+ */
+@Slf4j
+public class SouthTransmit extends AbstractTransmit {
+
+    private final IChannel channel;
+    private volatile Connection connection;
+    private volatile EFileTransfer eFileTransfer;
+    private volatile boolean isStop = false;
+
+    /** 连接配置 */
+    private String host;
+    private int port = 2404;
+    private int commonAddress = 1;
+    private int informationObjectAddress = 0;
+
+    public SouthTransmit(IChannel channel) {
+        this.channel = channel;
+    }
+
+    @Override
+    public int start(String config) {
+        Properties properties = new Properties();
+        try {
+            config = config.replace("\\n", "\n");
+            properties.load(new StringReader(config));
+            host = properties.getProperty("host");
+            if (host == null || host.isBlank()) {
+                throw new IllegalArgumentException("not exist host");
+            }
+            port = parseInt(properties, "port", 2404);
+            if (port == 0) {
+                throw new IllegalArgumentException("invalid port");
+            }
+            commonAddress = parseInt(properties, "commonAddress", 1);
+            informationObjectAddress = parseInt(properties, "informationObjectAddress", 0);
+            isStop = false;
+        } catch (Exception ex) {
+            log.error("channel={} config is invalid, content={}, ex={}", channel.getName(), config, ex);
+            return -1;
+        }
+        state = TransmitState.STARTED;
+        return 0;
+    }
+
+    @Override
+    public void stop() {
+        isStop = true;
+        super.stop();
+    }
+
+    @Override
+    public int connect() {
+        if (isStop) {
+            return -1;
+        }
+        try {
+            // EFileTransfer 作为连接监听器处理文件传输类 ASDU；其 delegate 留空，
+            // 报文上报改由注入的 MonitoringSocketFactory 在 socket 字节层切帧完成。
+            EFileTransfer transfer = new EFileTransfer(commonAddress, informationObjectAddress);
+
+            MonitoringSocketFactory socketFactory = new MonitoringSocketFactory(
+                    this::onFrameReceived, this::onFrameSent, this::onFrameClosed);
+
+            Connection conn = new ClientConnectionBuilder(InetAddress.getByName(host))
+                    .setPort(port)
+                    .setConnectionEventListener(transfer)
+                    .setSocketFactory(socketFactory)
+                    .build();
+            conn.startDataTransfer();
+            transfer.setConnection(conn);
+
+            connection = conn;
+            eFileTransfer = transfer;
+            notifyConnected();
+            log.info("channel={} connected to {}:{}", channel.getName(), host, port);
+            return 0;
+        } catch (Exception ex) {
+            notifyDisconnected();
+            log.warn("channel={} connect failed, ex={}", channel.getName(), ex.getMessage());
+            return -1;
+        }
+    }
+
+    @Override
+    public void disconnect() {
+        Connection conn = connection;
+        if (conn != null) {
+            try {
+                conn.close();
+            } catch (Exception ignored) {
+            }
+        }
+        connection = null;
+        eFileTransfer = null;
+        notifyDisconnected();
+    }
+
+    @Override
+    public int send(byte[] data) {
+        throw new UnsupportedOperationException("IEC 104 uses EFileTransfer, not raw byte send");
+    }
+
+    @Override
+    public int send(byte[] data, int tmOutMs) {
+        throw new UnsupportedOperationException("IEC 104 uses EFileTransfer, not raw byte send");
+    }
+
+    @Override
+    public byte[] receive() {
+        throw new UnsupportedOperationException("IEC 104 uses EFileTransfer, not raw byte receive");
+    }
+
+    @Override
+    public byte[] receive(int tmOutMs) {
+        throw new UnsupportedOperationException("IEC 104 uses EFileTransfer, not raw byte receive");
+    }
+
+    public Connection getConnection() {
+        return connection;
+    }
+
+    public EFileTransfer getFileTransfer() {
+        return eFileTransfer;
+    }
+
+    private static int parseInt(Properties p, String key, int def) {
+        String v = p.getProperty(key);
+        return (v == null || v.isBlank()) ? def : Integer.parseInt(v.trim());
+    }
+
+    /**
+     * 收到完整 APDU 帧（socket 字节层切帧）：触发 {@link #notifyReceived(byte[])}，
+     * 进入 {@link cn.oleauto.protocol.common.protocol.AbstractProtocol#onTransmitReceived}。
+     */
+    protected void onFrameReceived(byte[] frame) {
+        notifyReceived(frame);
+    }
+
+    /**
+     * 发出完整 APDU 帧（socket 字节层切帧）：触发 {@link #notifySent(byte[])}，
+     * 进入 {@link cn.oleauto.protocol.common.protocol.AbstractProtocol#onTransmitSent}。
+     */
+    protected void onFrameSent(byte[] frame) {
+        notifySent(frame);
+    }
+
+    /**
+     * 输入流 EOF 或 IOException：连接已断开，触发 {@link #notifyDisconnected()}。
+     */
+    protected void onFrameClosed() {
+        notifyDisconnected();
+        log.warn("channel={} socket stream closed", channel.getName());
+    }
+}

gateway/forecast-south/src/main/java/cn/oleauto/south/transfer/EFileTransfer.java

@@ -0,0 +1,539 @@
+package cn.oleauto.south.transfer;
+
+import cn.oleauto.iec60870.ASdu;
+import cn.oleauto.iec60870.ASduType;
+import cn.oleauto.iec60870.CauseOfTransmission;
+import cn.oleauto.iec60870.Connection;
+import cn.oleauto.iec60870.ConnectionEventListener;
+import cn.oleauto.iec60870.ie.IeAckFileOrSectionQualifier;
+import cn.oleauto.iec60870.ie.IeChecksum;
+import cn.oleauto.iec60870.ie.IeFileReadyQualifier;
+import cn.oleauto.iec60870.ie.IeFileSegment;
+import cn.oleauto.iec60870.ie.IeLastSectionOrSegmentQualifier;
+import cn.oleauto.iec60870.ie.IeLengthOfFileOrSection;
+import cn.oleauto.iec60870.ie.IeNameOfFile;
+import cn.oleauto.iec60870.ie.IeNameOfSection;
+import cn.oleauto.iec60870.ie.IeSectionReadyQualifier;
+import cn.oleauto.iec60870.ie.IeSelectAndCallQualifier;
+import cn.oleauto.iec60870.ie.InformationElement;
+import cn.oleauto.iec60870.ie.InformationObject;
+import cn.oleauto.south.efile.EDocument;
+import cn.oleauto.south.efile.EParser;
+import cn.oleauto.south.efile.EWriter;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+public class EFileTransfer implements ConnectionEventListener {
+
+    /** 默认每个分段的字节数（IEC 60870 ASDU 最大 253 字节，减去头部后约 240 字节，取 200 留足余量）。 */
+    public static final int DEFAULT_SEGMENT_SIZE = 200;
+
+    /** 默认等待对端响应超时（毫秒）。 */
+    public static final long DEFAULT_TIMEOUT_MS = 30_000L;
+
+    /** 文件传输过程中使用的公共地址。 */
+    private final int commonAddress;
+    private final int informationObjectAddress;
+    private final int segmentSize;
+    private final long timeoutMs;
+
+    private volatile Connection connection;
+    private volatile FileReceivedListener receivedListener;
+    private volatile ConnectionEventListener delegate;
+
+    // -------- 发送流程同步 --------
+    private volatile boolean sending;
+    private final LinkedBlockingQueue<ASdu> responseQueue = new LinkedBlockingQueue<>();
+
+    // -------- 接收流程状态 --------
+    private volatile boolean receiving;
+    private int receiveNameOfFile;
+    private int receiveExpectedLength;
+    private final ByteArrayOutputStream receiveBuffer = new ByteArrayOutputStream();
+    private int receiveChecksum;
+
+    /**
+     * 构造传输处理器（不绑定连接）。
+     *
+     * <p>适用于需要先把本对象注册为 {@code ConnectionEventListener}（例如通过
+     * {@code ClientConnectionBuilder.setConnectionEventListener(this)}）再构建连接的场景。
+     * 构建连接后需调用 {@link #setConnection(Connection)} 绑定。
+     *
+     * @param commonAddress           公共地址（ASDU Common Address）
+     * @param informationObjectAddress 信息对象地址
+     */
+    public EFileTransfer(int commonAddress, int informationObjectAddress) {
+        this(null, commonAddress, informationObjectAddress, DEFAULT_SEGMENT_SIZE, DEFAULT_TIMEOUT_MS);
+    }
+
+    /**
+     * 使用默认分段大小和超时时间构造。
+     *
+     * @param connection              已建立并调用 startDataTransfer() 的连接
+     * @param commonAddress           公共地址（ASDU Common Address）
+     * @param informationObjectAddress 信息对象地址
+     */
+    public EFileTransfer(Connection connection, int commonAddress, int informationObjectAddress) {
+        this(connection, commonAddress, informationObjectAddress,
+                DEFAULT_SEGMENT_SIZE, DEFAULT_TIMEOUT_MS);
+    }
+
+    public EFileTransfer(Connection connection, int commonAddress, int informationObjectAddress,
+                         int segmentSize, long timeoutMs) {
+        if (segmentSize < 1 || segmentSize > 240) {
+            throw new IllegalArgumentException("segmentSize 必须在 1~240 之间");
+        }
+        this.connection = connection;
+        this.commonAddress = commonAddress;
+        this.informationObjectAddress = informationObjectAddress;
+        this.segmentSize = segmentSize;
+        this.timeoutMs = timeoutMs;
+    }
+
+    /**
+     * 绑定连接。在用 {@link #EFileTransfer(int, int)} 构造后必须调用。
+     *
+     * @param connection 已建立的连接
+     */
+    public void setConnection(Connection connection) {
+        this.connection = connection;
+    }
+
+    // ====================================================================
+    //  公开 API —— 发送
+    // ====================================================================
+
+    /**
+     * 发送一个 E 语言文档（先序列化为字节，再走 IEC 60870 文件传输）。
+     *
+     * @param nameOfFile 文件名（IEC 60870 NOF，1~65535）
+     * @param document   E 语言文档
+     * @param charset    序列化字符集（常用 UTF-8 或 GB18030）
+     */
+    public void sendFile(int nameOfFile, EDocument document, Charset charset) throws IOException {
+        if (document == null) {
+            throw new IllegalArgumentException("document 不能为空");
+        }
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        new EWriter().write(document, bos, charset);
+        sendFile(nameOfFile, bos.toByteArray());
+    }
+
+    /**
+     * 发送磁盘上的 E 语言文件（按原始字节传输，不解析）。
+     */
+    public void sendFile(int nameOfFile, File file) throws IOException {
+        if (file == null || !file.isFile()) {
+            throw new IllegalArgumentException("file 不存在或不是文件: " + file);
+        }
+        sendFile(nameOfFile, Files.readAllBytes(file.toPath()));
+    }
+
+    /**
+     * 发送原始字节（接收方会尝试按 E 语言解析，解析失败时回调中 {@code document} 为 null）。
+     *
+     * @param nameOfFile 文件名（NOF）
+     * @param data       文件内容
+     */
+    public void sendFile(int nameOfFile, byte[] data) throws IOException {
+        if (data == null) {
+            throw new IllegalArgumentException("data 不能为空");
+        }
+        if (connection == null) {
+            throw new IllegalStateException("未绑定 Connection，请先调用 setConnection()");
+        }
+        if (sending || receiving) {
+            throw new IllegalStateException("已有文件传输正在进行");
+        }
+
+        sending = true;
+        try {
+            doSend(nameOfFile, data);
+        } finally {
+            sending = false;
+            responseQueue.clear();
+        }
+    }
+
+    // ====================================================================
+    //  公开 API —— 接收回调
+    // ====================================================================
+
+    public void setFileReceivedListener(FileReceivedListener listener) {
+        this.receivedListener = listener;
+    }
+
+    /**
+     * 设置非文件传输 ASDU 的委托监听器。文件传输相关的 ASDU 不会被透传给委托。
+     */
+    public void setDelegate(ConnectionEventListener delegate) {
+        this.delegate = delegate;
+    }
+
+    // ====================================================================
+    //  ConnectionEventListener 实现
+    // ====================================================================
+
+    @Override
+    public void newASdu(Connection connection, ASdu aSdu) {
+        ASduType type = aSdu.getTypeIdentification();
+
+        // 发送流程中：把文件传输响应放进队列，供 sendFile 的 wait 方法的消费
+        if (sending && isFileTransferType(type)) {
+            responseQueue.offer(aSdu);
+            return;
+        }
+
+        // 接收流程：处理入站文件传输 ASDU
+        if (isFileTransferType(type)) {
+            try {
+                handleReceiveFlow(connection, aSdu);
+            } catch (Exception e) {
+                notifyError(receiveNameOfFile, "接收文件传输 ASDU 失败: " + e.getMessage());
+                receiving = false;
+                receiveBuffer.reset();
+            }
+            return;
+        }
+
+        // 非文件传输 ASDU 透传
+        if (delegate != null) {
+            delegate.newASdu(connection, aSdu);
+        }
+    }
+
+    @Override
+    public void connectionClosed(Connection connection, IOException cause) {
+        sending = false;
+        receiving = false;
+        responseQueue.clear();
+        if (delegate != null) {
+            delegate.connectionClosed(connection, cause);
+        }
+    }
+
+    @Override
+    public void dataTransferStateChanged(Connection connection, boolean stopped) {
+        if (delegate != null) {
+            delegate.dataTransferStateChanged(connection, stopped);
+        }
+    }
+
+    // ====================================================================
+    //  发送流程实现
+    // ====================================================================
+
+    private void doSend(int nameOfFile, byte[] data) throws IOException {
+        int checksum = computeChecksum(data);
+
+        // 1. fileReady
+        connection.fileReady(commonAddress, informationObjectAddress,
+                new IeNameOfFile(nameOfFile),
+                new IeLengthOfFileOrSection(data.length),
+                new IeFileReadyQualifier(1, false));
+
+        // 2. 等待 callOrSelectFiles (action=2 = call file)
+        ASdu call = waitForResponse("call file");
+        if (!isCallFile(call, nameOfFile)) {
+            throw new IOException("期望 call file 响应，实际收到: " + call.getTypeIdentification());
+        }
+
+        // 3. sectionReady (section 0)
+        connection.sectionReady(commonAddress, informationObjectAddress,
+                new IeNameOfFile(nameOfFile),
+                new IeNameOfSection(0),
+                new IeLengthOfFileOrSection(data.length),
+                new IeSectionReadyQualifier(1, false));
+
+        // 4. 等待 ack section (action=2)
+        ASdu ackSection = waitForResponse("ack section");
+        if (!isAckSection(ackSection)) {
+            throw new IOException("期望 ack section 响应，实际收到: " + ackSection.getTypeIdentification());
+        }
+
+        // 5. 发送分段
+        int offset = 0;
+        while (offset < data.length) {
+            int len = Math.min(segmentSize, data.length - offset);
+            byte[] seg = new byte[len];
+            System.arraycopy(data, offset, seg, 0, len);
+            offset += len;
+
+            connection.sendSegment(commonAddress, informationObjectAddress,
+                    new IeNameOfFile(nameOfFile),
+                    new IeNameOfSection(0),
+                    new IeFileSegment(seg, 0, len));
+        }
+
+        // 6. lastSectionOrSegment (qualifier=3 = last segment of file) + checksum
+        connection.lastSectionOrSegment(commonAddress, informationObjectAddress,
+                new IeNameOfFile(nameOfFile),
+                new IeNameOfSection(0),
+                new IeLastSectionOrSegmentQualifier(3),
+                new IeChecksum(checksum));
+
+        // 7. 等待 ack section (action=2)
+        ASdu ackSec2 = waitForResponse("ack section (after last segment)");
+        if (!isAckSection(ackSec2)) {
+            throw new IOException("期望末段后 ack section，实际收到: " + ackSec2.getTypeIdentification());
+        }
+
+        // 8. 等待 ack file (action=1)
+        ASdu ackFile = waitForResponse("ack file");
+        if (!isAckFile(ackFile)) {
+            throw new IOException("期望 ack file 响应，实际收到: " + ackFile.getTypeIdentification());
+        }
+    }
+
+    private ASdu waitForResponse(String what) throws IOException {
+        try {
+            ASdu response = responseQueue.poll(timeoutMs, TimeUnit.MILLISECONDS);
+            if (response == null) {
+                throw new IOException("等待 " + what + " 超时（" + timeoutMs + " ms）");
+            }
+            return response;
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("等待 " + what + " 被中断", e);
+        }
+    }
+
+    // ====================================================================
+    //  接收流程实现
+    // ====================================================================
+
+    private void handleReceiveFlow(Connection connection, ASdu aSdu) throws IOException {
+        ASduType type = aSdu.getTypeIdentification();
+        InformationObject io = firstInformationObject(aSdu);
+        if (io == null) {
+            return;
+        }
+
+        switch (type) {
+            case F_FR_NA_1: // 120 fileReady
+                onFileReady(connection, io);
+                break;
+            case F_SR_NA_1: // 121 sectionReady
+                onSectionReady(connection, io);
+                break;
+            case F_SG_NA_1: // 125 segment
+                onSegment(io);
+                break;
+            case F_LS_NA_1: // 123 lastSectionOrSegment
+                onLastSectionOrSegment(connection, io);
+                break;
+            default:
+                // 其他文件传输类型（如 F_SC_NA_1 call/select、F_AF_NA_1 ack）在接收方不主动处理
+                break;
+        }
+    }
+
+    private void onFileReady(Connection connection, InformationObject io) throws IOException {
+        InformationElement[] elems = firstElementSet(io);
+        IeNameOfFile nof = (IeNameOfFile) elems[0];
+        IeLengthOfFileOrSection lof = (IeLengthOfFileOrSection) elems[1];
+
+        receiveNameOfFile = nof.getValue();
+        receiveExpectedLength = lof.getValue();
+        receiveBuffer.reset();
+        receiveChecksum = 0;
+        receiving = true;
+
+        // 回复 call file (SCQ action=2)
+        connection.callOrSelectFiles(commonAddress, CauseOfTransmission.FILE_TRANSFER,
+                informationObjectAddress,
+                new IeNameOfFile(receiveNameOfFile),
+                new IeNameOfSection(0),
+                new IeSelectAndCallQualifier(2, 0));
+    }
+
+    private void onSectionReady(Connection connection, InformationObject io) throws IOException {
+        InformationElement[] elems = firstElementSet(io);
+        IeNameOfFile nof = (IeNameOfFile) elems[0];
+        IeNameOfSection nos = (IeNameOfSection) elems[1];
+        IeLengthOfFileOrSection los = (IeLengthOfFileOrSection) elems[2];
+
+        // 回复 ack section (AFQ action=2)
+        connection.ackFileOrSection(commonAddress, informationObjectAddress,
+                new IeNameOfFile(nof.getValue()),
+                nos,
+                new IeAckFileOrSectionQualifier(2, 0));
+    }
+
+    private void onSegment(InformationObject io) throws IOException {
+        InformationElement[] elems = firstElementSet(io);
+        IeFileSegment segment = (IeFileSegment) elems[2];
+        byte[] data = segment.getSegment();
+        receiveBuffer.write(data);
+        for (byte b : data) {
+            receiveChecksum = (receiveChecksum + (b & 0xff)) & 0xff;
+        }
+    }
+
+    private void onLastSectionOrSegment(Connection connection, InformationObject io) throws IOException {
+        InformationElement[] elems = firstElementSet(io);
+        IeNameOfFile nof = (IeNameOfFile) elems[0];
+        IeNameOfSection nos = (IeNameOfSection) elems[1];
+        IeLastSectionOrSegmentQualifier lsq = (IeLastSectionOrSegmentQualifier) elems[2];
+        IeChecksum chs = (IeChecksum) elems[3];
+
+        byte[] fileData = receiveBuffer.toByteArray();
+        int expectedChecksum = computeChecksum(fileData);
+
+        // ack section (action=2)
+        connection.ackFileOrSection(commonAddress, informationObjectAddress,
+                new IeNameOfFile(nof.getValue()),
+                nos,
+                new IeAckFileOrSectionQualifier(2, 0));
+
+        // ack file (action=1)
+        connection.ackFileOrSection(commonAddress, informationObjectAddress,
+                new IeNameOfFile(nof.getValue()),
+                nos,
+                new IeAckFileOrSectionQualifier(1, 0));
+
+        receiving = false;
+
+        // 校验校验和
+        if (chs.getValue() != expectedChecksum) {
+            notifyError(nof.getValue(),
+                    "校验和不匹配: 收到=" + chs.getValue() + ", 计算=" + expectedChecksum);
+            return;
+        }
+
+        // 尝试解析为 E 语言文档
+        EDocument document = null;
+        try {
+            document = new EParser().parse(new String(fileData, StandardCharsets.UTF_8));
+        } catch (Exception ignored) {
+            // 非 E 语言或 GBK 编码时尝试 GB18030
+            try {
+                document = new EParser().parse(new String(fileData,
+                        cn.oleauto.south.efile.ECharset.GB18030));
+            } catch (Exception ignored2) {
+                // 确实不是 E 语言文件，回调中 document 为 null
+            }
+        }
+
+        // 回调
+        FileReceivedListener listener = receivedListener;
+        if (listener != null) {
+            listener.onFileReceived(nof.getValue(), fileData, document);
+        }
+
+        // 重置接收缓冲
+        receiveBuffer.reset();
+        receiveChecksum = 0;
+    }
+
+    // ====================================================================
+    //  工具方法
+    // ====================================================================
+
+    private boolean isFileTransferType(ASduType type) {
+        switch (type) {
+            case F_FR_NA_1:
+            case F_SR_NA_1:
+            case F_SC_NA_1:
+            case F_LS_NA_1:
+            case F_AF_NA_1:
+            case F_SG_NA_1:
+                return true;
+            default:
+                return false;
+        }
+    }
+
+    private boolean isCallFile(ASdu aSdu, int nameOfFile) {
+        if (aSdu.getTypeIdentification() != ASduType.F_SC_NA_1) {
+            return false;
+        }
+        InformationObject io = firstInformationObject(aSdu);
+        if (io == null) {
+            return false;
+        }
+        InformationElement[] elems = firstElementSet(io);
+        IeNameOfFile nof = (IeNameOfFile) elems[0];
+        IeSelectAndCallQualifier scq = (IeSelectAndCallQualifier) elems[2];
+        return nof.getValue() == nameOfFile && scq.getRequest() == 2;
+    }
+
+    private boolean isAckSection(ASdu aSdu) {
+        return isAckOf(aSdu, 2);
+    }
+
+    private boolean isAckFile(ASdu aSdu) {
+        return isAckOf(aSdu, 1);
+    }
+
+    private boolean isAckOf(ASdu aSdu, int action) {
+        if (aSdu.getTypeIdentification() != ASduType.F_AF_NA_1) {
+            return false;
+        }
+        InformationObject io = firstInformationObject(aSdu);
+        if (io == null) {
+            return false;
+        }
+        InformationElement[] elems = firstElementSet(io);
+        IeAckFileOrSectionQualifier afq = (IeAckFileOrSectionQualifier) elems[2];
+        return afq.getRequest() == action;
+    }
+
+    private InformationObject firstInformationObject(ASdu aSdu) {
+        InformationObject[] ios = aSdu.getInformationObjects();
+        return (ios != null && ios.length > 0) ? ios[0] : null;
+    }
+
+    private InformationElement[] firstElementSet(InformationObject io) {
+        InformationElement[][] all = io.getInformationElements();
+        return (all != null && all.length > 0) ? all[0] : new InformationElement[0];
+    }
+
+    private static int computeChecksum(byte[] data) {
+        int sum = 0;
+        for (byte b : data) {
+            sum = (sum + (b & 0xff)) & 0xff;
+        }
+        return sum;
+    }
+
+    private void notifyError(int nameOfFile, String message) {
+        FileReceivedListener listener = receivedListener;
+        if (listener != null) {
+            listener.onError(nameOfFile, message);
+        }
+    }
+
+    // ====================================================================
+    //  回调接口
+    // ====================================================================
+
+    /**
+     * 文件接收完成回调。
+     */
+    public interface FileReceivedListener {
+
+        /**
+         * 文件接收完成时调用。
+         *
+         * @param nameOfFile 文件名（NOF）
+         * @param data       原始字节
+         * @param document   解析后的 E 语言文档；如果数据不是合法 E 语言文件则为 null
+         */
+        void onFileReceived(int nameOfFile, byte[] data, EDocument document);
+
+        /**
+         * 文件传输出错时调用（校验和失败、协议异常等）。
+         */
+        default void onError(int nameOfFile, String message) {
+            // 默认空实现，子类按需覆写
+        }
+    }
+}

gateway/forecast-south/src/main/java/cn/oleauto/south/transfer/EFileTransferDemo.java

@@ -0,0 +1,240 @@
+package cn.oleauto.south.transfer;
+
+import cn.hutool.core.thread.ThreadUtil;
+import cn.oleauto.iec60870.Connection;
+import cn.oleauto.iec60870.ConnectionEventListener;
+import cn.oleauto.iec60870.Server;
+import cn.oleauto.iec60870.ServerEventListener;
+import cn.oleauto.south.efile.EBlock;
+import cn.oleauto.south.efile.EDocument;
+import cn.oleauto.south.efile.EParser;
+import cn.oleauto.south.efile.EWriter;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 演示：使用 IEC 60870-5-104 文件传输服务发送和接收 E 语言文件。
+ *
+ * <p>启动一个本地 IEC 104 服务端和一个客户端，客户端构造一份 E 语言文档，
+ * 通过 {@link EFileTransfer} 发送，服务端收到后自动解析并打印。
+ * 怎么启动，怎么串起来
+ */
+public class EFileTransferDemo {
+
+    /**
+     * 报文序列如下
+     buffer=:680407000000
+     buffer=:68040b000000
+     buffer=:68130000000078010d0001000000000100fd000001
+     buffer=:6811000002007a010d00010000000001000002
+     buffer=:68140200020079010d000100000000010000fd000001
+     buffer=:6811020004007c010d00010000000001000002
+     buffer=:68d9040004007d010d000100000000010000c83c212053797374656d3d4f4d532056657273696f6e3d312e3020436f64653d5445535420446174613d323032362d30392d313720213e0d0a3c506f7765725472616e73666f726d65723e0d0a4009e58f98e58e8be599a8e5908d09e58e82e7ab99e5908d09e7b1bbe59e8b09e9a29de5ae9ae58a9fe78e870d0a2509530953095309460d0a24092d092d092d094d56410d0a3a092d092d092d09303a393939390d0a2309e4b8bbe58f983109e58d8ee4b89ce7ab990932e58db7e58f98093138302e300d0a2309e4
+     buffer=:6846060004007d010d00010000000001000035b8bbe58f983209e58d8ee4b89ce7ab990933e58db7e58f98093234302e300d0a3c2f506f7765725472616e73666f726d65723e0d0a
+     buffer=:6812080004007b010d00010000000001000003b2
+     buffer=:681104000a007c010d00010000000001000002
+     buffer=:681106000a007c010d00010000000001000001
+     */
+    public static void main2(String[] args) throws Exception {
+        int port = 13098;
+        int commonAddress = 1;
+        int ioa = 0;
+
+        // -------- 构造一份 E 语言文档 --------
+        EDocument doc = buildSampleDocument();
+
+        // -------- 服务端（接收方） --------
+        Server server = Server.builder()
+                .setPort(port)
+                .setBindAddr(InetAddress.getByName("127.0.0.1"))
+                .build();
+
+        CountDownLatch received = new CountDownLatch(1);
+        int[] receivedLength = {0};
+
+        server.start(new ServerEventListener() {
+            @Override
+            public ConnectionEventListener connectionIndication(Connection connection) {
+                EFileTransfer transfer = new EFileTransfer(connection, commonAddress, ioa);
+                transfer.setFileReceivedListener((nameOfFile, data, document) -> {
+                    receivedLength[0] = data.length;
+                    System.out.println("===== 服务端收到文件 =====");
+                    System.out.println("文件名(NOF): " + nameOfFile);
+                    System.out.println("字节数: " + data.length);
+                    if (document != null) {
+                        System.out.println("E 语言块数: " + document.getEBlocks().size());
+                        EBlock block = document.getEBlocks().get(0);
+                        System.out.println("类名: " + block.getClassName());
+                        System.out.println("列名: " + block.getColumns());
+                        System.out.println("数据行数: " + block.getRows().size());
+                        System.out.println("第 1 行: " + block.getRows().get(0));
+                    } else {
+                        System.out.println("非 E 语言文档，原始内容:");
+                        System.out.println(new String(data, StandardCharsets.UTF_8));
+                    }
+                    received.countDown();
+                });
+                return transfer;
+            }
+
+            @Override
+            public void serverStoppedListeningIndication(IOException e) {
+                System.err.println("服务端停止监听: " + e.getMessage());
+            }
+
+            @Override
+            public void connectionAttemptFailed(IOException e) {
+                System.err.println("连接尝试失败: " + e.getMessage());
+            }
+        });
+
+        // 给服务端一点时间启动
+        Thread.sleep(500);
+
+        // -------- 客户端（发送方） --------
+        // 先创建 EFileTransfer（不带连接），注册为 ConnectionEventListener，
+        // 构建 Connection 后再绑定，确保 server 回复的 ASDU 能被本对象接收。
+        EFileTransfer sender = new EFileTransfer(commonAddress, ioa);
+        Connection connection = new cn.oleauto.iec60870.ClientConnectionBuilder("127.0.0.1")
+                .setPort(port)
+                .setConnectionEventListener(sender)
+                .build();
+        sender.setConnection(connection);
+        connection.startDataTransfer();
+
+        System.out.println("===== 客户端发送 E 语言文件 =====");
+        System.out.println("块数: " + doc.getEBlocks().size());
+
+        sender.sendFile(1, doc, StandardCharsets.UTF_8);
+        System.out.println("发送完成");
+
+        // 等待服务端收到
+        if (!received.await(10, TimeUnit.SECONDS)) {
+            System.err.println("等待接收超时");
+        } else {
+            System.out.println("接收确认，字节数=" + receivedLength[0]);
+        }
+
+        // 清理
+        connection.close();
+        server.stop();
+        System.out.println("演示结束");
+    }
+
+    public static void main(String[] args) throws Exception {
+        int port = 13098;
+        int commonAddress = 1;
+        int ioa = 0;
+
+        // -------- 构造一份 E 语言文档 --------
+        EDocument doc = buildSampleDocument();
+
+        // -------- 服务端（接收方） --------
+        Server server = Server.builder()
+                .setPort(port)
+                .setBindAddr(InetAddress.getByName("127.0.0.1"))
+                .build();
+
+        CountDownLatch received = new CountDownLatch(1);
+        int[] receivedLength = {0};
+
+        server.start(new ServerEventListener() {
+            @Override
+            public ConnectionEventListener connectionIndication(Connection connection) {
+                EFileTransfer transfer = new EFileTransfer(connection, commonAddress, ioa);
+                transfer.setFileReceivedListener((nameOfFile, data, document) -> {
+                    receivedLength[0] = data.length;
+                    System.out.println("===== 服务端收到文件 =====");
+                    System.out.println("文件名(NOF): " + nameOfFile);
+                    System.out.println("字节数: " + data.length);
+                    if (document != null) {
+                        System.out.println("E 语言块数: " + document.getEBlocks().size());
+                        EBlock block = document.getEBlocks().get(0);
+                        System.out.println("类名: " + block.getClassName());
+                        System.out.println("列名: " + block.getColumns());
+                        System.out.println("数据行数: " + block.getRows().size());
+                        System.out.println("第 1 行: " + block.getRows().get(0));
+                    } else {
+                        System.out.println("非 E 语言文档，原始内容:");
+                        System.out.println(new String(data, StandardCharsets.UTF_8));
+                    }
+                    received.countDown();
+                });
+                return transfer;
+            }
+
+            @Override
+            public void serverStoppedListeningIndication(IOException e) {
+                System.err.println("服务端停止监听: " + e.getMessage());
+            }
+
+            @Override
+            public void connectionAttemptFailed(IOException e) {
+                System.err.println("连接尝试失败: " + e.getMessage());
+            }
+        });
+
+        // 给服务端一点时间启动
+        Thread.sleep(500);
+
+        // 等待服务端收到
+//        if (!received.await(10, TimeUnit.SECONDS)) {
+//            System.err.println("等待接收超时");
+//        } else {
+//            System.out.println("接收确认，字节数=" + receivedLength[0]);
+//        }
+//        // 清理
+//        server.stop();
+//        System.out.println("演示结束");
+
+        // 测试块
+        while(!received.await(10, TimeUnit.SECONDS)){
+            ThreadUtil.sleep(1000);
+        }
+        System.out.println("接收确认，字节数=" + receivedLength[0]);
+
+    }
+
+    public static EDocument buildSampleDocument() {
+        EDocument doc = new EDocument();
+        doc.setDeclaration("System=OMS Version=1.0 Code=TEST Data=2026-09-17");
+
+        EBlock block = new EBlock("PowerTransformer");
+        block.setColumns(Arrays.asList("变压器名", "厂站名", "类型", "额定功率"));
+        block.setTypes(Arrays.asList("S", "S", "S", "F"));
+        block.setUnits(Arrays.asList("-", "-", "-", "MVA"));
+        block.setLimits(Arrays.asList("-", "-", "-", "0:9999"));
+
+        Map<String, String> row1 = new LinkedHashMap<>();
+        row1.put("变压器名", "主变1");
+        row1.put("厂站名", "华东站");
+        row1.put("类型", "2卷变");
+        row1.put("额定功率", "180.0");
+        block.addRow(row1);
+
+        Map<String, String> row2 = new LinkedHashMap<>();
+        row2.put("变压器名", "主变2");
+        row2.put("厂站名", "华东站");
+        row2.put("类型", "3卷变");
+        row2.put("额定功率", "240.0");
+        block.addRow(row2);
+
+        doc.addEBlock(block);
+
+        // 打印发送的 E 语言文本
+        java.io.ByteArrayOutputStream bos = new java.io.ByteArrayOutputStream();
+        try {
+            new EWriter().write(doc, bos, StandardCharsets.UTF_8);
+            System.out.println("E 语言文本:\n" + new String(bos.toByteArray(), StandardCharsets.UTF_8));
+        } catch (IOException ignored) {
+        }
+        return doc;
+    }
+}

gateway/forecast-south/src/test/java/cn/oleauto/south/ApduSlicerTest.java

@@ -0,0 +1,148 @@
+package cn.oleauto.south;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * {@link ApduSlicer} 单元测试，覆盖单帧/多帧、跨 chunk 拼接、重新同步、长度边界与体内 0x68。
+ *
+ * <p>帧格式：0x68 + L(4..253) + L 字节体，总长 2+L。
+ */
+public class ApduSlicerTest {
+
+    private static final byte[] TESTFR_ACT = {0x68, 0x04, 0x43, 0x00, 0x00, 0x00};
+    private static final byte[] STARTDT_ACT = {0x68, 0x04, 0x07, 0x00, 0x00, 0x00};
+
+    private static List<byte[]> feed(ApduSlicer slicer, byte[] data) {
+        return feed(slicer, data, 0, data.length);
+    }
+
+    private static List<byte[]> feed(ApduSlicer slicer, byte[] data, int off, int len) {
+        List<byte[]> out = new ArrayList<>();
+        slicer.feed(data, off, len, out::add);
+        return out;
+    }
+
+    private static void assertFrames(List<byte[]> actual, byte[]... expected) {
+        assertEquals("帧数量", expected.length, actual.size());
+        for (int i = 0; i < expected.length; i++) {
+            assertArrayEquals("第 " + i + " 帧内容", expected[i], actual.get(i));
+        }
+    }
+
+    @Test
+    public void testSingleCompleteFrame() {
+        ApduSlicer slicer = new ApduSlicer();
+        assertFrames(feed(slicer, TESTFR_ACT), TESTFR_ACT);
+    }
+
+    @Test
+    public void testMultipleFramesInOneChunk() {
+        ApduSlicer slicer = new ApduSlicer();
+        byte[] chunk = concat(TESTFR_ACT, STARTDT_ACT);
+        assertFrames(feed(slicer, chunk), TESTFR_ACT, STARTDT_ACT);
+    }
+
+    @Test
+    public void testFrameSplitAcrossFeeds() {
+        ApduSlicer slicer = new ApduSlicer();
+        assertTrue(feed(slicer, Arrays.copyOf(TESTFR_ACT, 3)).isEmpty()); // 0x68,0x04,0x43
+        assertFrames(feed(slicer, rest(TESTFR_ACT, 3)), TESTFR_ACT);
+    }
+
+    @Test
+    public void testSplitRightAfterStartByte() {
+        ApduSlicer slicer = new ApduSlicer();
+        assertTrue(feed(slicer, new byte[]{0x68}).isEmpty());
+        assertFrames(feed(slicer, new byte[]{0x04, 0x43, 0x00, 0x00, 0x00}), TESTFR_ACT);
+    }
+
+    @Test
+    public void testFrameAcrossThreeFeeds() {
+        ApduSlicer slicer = new ApduSlicer();
+        assertTrue(feed(slicer, new byte[]{0x68, 0x04}).isEmpty());
+        assertTrue(feed(slicer, new byte[]{0x43, 0x00}).isEmpty());
+        assertFrames(feed(slicer, new byte[]{0x00, 0x00}), TESTFR_ACT);
+    }
+
+    @Test
+    public void testLeadingGarbageThenFrame() {
+        ApduSlicer slicer = new ApduSlicer();
+        byte[] chunk = {0x00, 0x11, 0x68, 0x04, 0x43, 0x00, 0x00, 0x00};
+        assertFrames(feed(slicer, chunk), TESTFR_ACT);
+    }
+
+    @Test
+    public void testInvalidLengthTooSmallResync() {
+        ApduSlicer slicer = new ApduSlicer();
+        byte[] chunk = {0x68, 0x03, 0x68, 0x04, 0x43, 0x00, 0x00, 0x00};
+        assertFrames(feed(slicer, chunk), TESTFR_ACT);
+    }
+
+    @Test
+    public void testInvalidLengthTooLargeResync() {
+        ApduSlicer slicer = new ApduSlicer();
+        byte[] chunk = {0x68, (byte) 0xFE, 0x68, 0x04, 0x43, 0x00, 0x00, 0x00};
+        assertFrames(feed(slicer, chunk), TESTFR_ACT);
+    }
+
+    @Test
+    public void testEmptyInput() {
+        ApduSlicer slicer = new ApduSlicer();
+        assertTrue(feed(slicer, new byte[0]).isEmpty());
+    }
+
+    @Test
+    public void testMaxLengthBoundary() {
+        ApduSlicer slicer = new ApduSlicer();
+        byte[] frame = buildFrame(253); // L=253, 总长 255
+        assertFrames(feed(slicer, frame), frame);
+    }
+
+    @Test
+    public void testBodyContainingStartByte() {
+        ApduSlicer slicer = new ApduSlicer();
+        // 帧体首字节为 0x68：正常传输下 emit 后 pos 跳过整帧，不应误判体内 0x68 为起始
+        byte[] frame = {0x68, 0x05, 0x68, 0x00, (byte) 0xAA, (byte) 0xBB, (byte) 0xCC};
+        assertFrames(feed(slicer, frame), frame);
+    }
+
+    @Test
+    public void testTwoFramesBodyEndsWithStartByte() {
+        ApduSlicer slicer = new ApduSlicer();
+        // 第一帧体最后一字节为 0x68，紧接第二帧，验证 emit 后 pos 正确落到第二帧边界
+        byte[] frame1 = {0x68, 0x05, 0x00, 0x00, 0x00, 0x00, 0x68};
+        byte[] chunk = concat(frame1, TESTFR_ACT);
+        assertFrames(feed(slicer, chunk), frame1, TESTFR_ACT);
+    }
+
+    private static byte[] concat(byte[] a, byte[] b) {
+        byte[] out = new byte[a.length + b.length];
+        System.arraycopy(a, 0, out, 0, a.length);
+        System.arraycopy(b, 0, out, a.length, b.length);
+        return out;
+    }
+
+    private static byte[] rest(byte[] src, int from) {
+        byte[] out = new byte[src.length - from];
+        System.arraycopy(src, from, out, 0, out.length);
+        return out;
+    }
+
+    private static byte[] buildFrame(int bodyLen) {
+        byte[] frame = new byte[2 + bodyLen];
+        frame[0] = 0x68;
+        frame[1] = (byte) bodyLen;
+        for (int i = 0; i < bodyLen; i++) {
+            frame[2 + i] = (byte) (i & 0xFF);
+        }
+        return frame;
+    }
+}

gateway/forecast-south/src/test/java/cn/oleauto/south/MonitoringInputStreamTest.java

@@ -0,0 +1,113 @@
+package cn.oleauto.south;
+
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * {@link MonitoringInputStream} 单元测试：切帧回调、EOF/IOException 触发 onClosed 且仅一次、单字节读路径。
+ */
+public class MonitoringInputStreamTest {
+
+    private static final byte[] TESTFR_ACT = {0x68, 0x04, 0x43, 0x00, 0x00, 0x00};
+
+    @Test
+    public void testBulkReadEmitsFrame() throws IOException {
+        List<byte[]> frames = new ArrayList<>();
+        int[] closed = {0};
+        MonitoringInputStream in = new MonitoringInputStream(
+                new ByteArrayInputStream(TESTFR_ACT), frames::add, () -> closed[0]++);
+
+        byte[] buf = new byte[TESTFR_ACT.length];
+        int n = in.read(buf, 0, buf.length);
+        assertEquals(TESTFR_ACT.length, n);
+        assertArrayEquals(TESTFR_ACT, buf);
+        assertEquals(1, frames.size());
+        assertArrayEquals(TESTFR_ACT, frames.get(0));
+        assertEquals(0, closed[0]);
+    }
+
+    @Test
+    public void testEofTriggersOnClosedOnce() throws IOException {
+        List<byte[]> frames = new ArrayList<>();
+        int[] closed = {0};
+        MonitoringInputStream in = new MonitoringInputStream(
+                new ByteArrayInputStream(TESTFR_ACT), frames::add, () -> closed[0]++);
+
+        byte[] buf = new byte[16];
+        in.read(buf, 0, 16); // 读出全部 6 字节
+        assertEquals(1, frames.size());
+        assertEquals(0, closed[0]);
+
+        assertEquals(-1, in.read(buf, 0, 16)); // EOF
+        assertEquals(1, closed[0]);
+
+        assertEquals(-1, in.read(buf, 0, 16)); // 再次 EOF
+        assertEquals("onClosed 不应重复触发", 1, closed[0]);
+    }
+
+    @Test
+    public void testSingleByteReadEmitsFrame() throws IOException {
+        List<byte[]> frames = new ArrayList<>();
+        MonitoringInputStream in = new MonitoringInputStream(
+                new ByteArrayInputStream(TESTFR_ACT), frames::add, null);
+
+        int b;
+        while ((b = in.read()) != -1) {
+            assertTrue(b >= 0 && b <= 0xFF);
+        }
+        assertEquals(1, frames.size());
+        assertArrayEquals(TESTFR_ACT, frames.get(0));
+    }
+
+    @Test
+    public void testChunkedReadReassemblesFrame() throws IOException {
+        // 模拟网络分片：每次读 2 字节，slicer 跨次拼接
+        List<byte[]> frames = new ArrayList<>();
+        MonitoringInputStream in = new MonitoringInputStream(
+                new ByteArrayInputStream(TESTFR_ACT), frames::add, null);
+
+        byte[] buf = new byte[2];
+        for (int i = 0; i < 3; i++) {
+            assertEquals(2, in.read(buf, 0, 2));
+        }
+        assertEquals(1, frames.size());
+        assertArrayEquals(TESTFR_ACT, frames.get(0));
+    }
+
+    @Test
+    public void testIOExceptionTriggersOnClosed() {
+        int[] closed = {0};
+        MonitoringInputStream in = new MonitoringInputStream(
+                new ThrowingInputStream(), null, () -> closed[0]++);
+        try {
+            in.read(new byte[4], 0, 4);
+            fail("应抛 IOException");
+        } catch (IOException e) {
+            assertEquals("boom", e.getMessage());
+        }
+        assertEquals(1, closed[0]);
+    }
+
+    /** 读到即抛 IOException 的输入流，验证 onClosed 触发。 */
+    private static class ThrowingInputStream extends InputStream {
+        @Override
+        public int read() throws IOException {
+            throw new IOException("boom");
+        }
+
+        @Override
+        public int read(byte[] b, int off, int len) throws IOException {
+            throw new IOException("boom");
+        }
+    }
+}

gateway/forecast-south/src/test/java/cn/oleauto/south/MonitoringOutputStreamTest.java

@@ -0,0 +1,100 @@
+package cn.oleauto.south;
+
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * {@link MonitoringOutputStream} 单元测试：切帧回调、字节透传、单字节写、分片写、flush 不重复。
+ */
+public class MonitoringOutputStreamTest {
+
+    private static final byte[] TESTFR_ACT = {0x68, 0x04, 0x43, 0x00, 0x00, 0x00};
+    private static final byte[] STARTDT_ACT = {0x68, 0x04, 0x07, 0x00, 0x00, 0x00};
+
+    @Test
+    public void testBulkWriteEmitsFrameAndPassesThrough() throws IOException {
+        ByteArrayOutputStream sink = new ByteArrayOutputStream();
+        List<byte[]> frames = new ArrayList<>();
+        MonitoringOutputStream out = new MonitoringOutputStream(sink, frames::add);
+
+        out.write(TESTFR_ACT, 0, TESTFR_ACT.length);
+
+        assertArrayEquals(TESTFR_ACT, sink.toByteArray());
+        assertEquals(1, frames.size());
+        assertArrayEquals(TESTFR_ACT, frames.get(0));
+    }
+
+    @Test
+    public void testSingleByteWriteReassemblesFrame() throws IOException {
+        ByteArrayOutputStream sink = new ByteArrayOutputStream();
+        List<byte[]> frames = new ArrayList<>();
+        MonitoringOutputStream out = new MonitoringOutputStream(sink, frames::add);
+
+        for (byte b : TESTFR_ACT) {
+            out.write(b & 0xFF);
+        }
+
+        assertArrayEquals(TESTFR_ACT, sink.toByteArray());
+        assertEquals(1, frames.size());
+        assertArrayEquals(TESTFR_ACT, frames.get(0));
+    }
+
+    @Test
+    public void testTwoFramesInOneWrite() throws IOException {
+        ByteArrayOutputStream sink = new ByteArrayOutputStream();
+        List<byte[]> frames = new ArrayList<>();
+        MonitoringOutputStream out = new MonitoringOutputStream(sink, frames::add);
+
+        byte[] chunk = concat(TESTFR_ACT, STARTDT_ACT);
+        out.write(chunk, 0, chunk.length);
+
+        assertArrayEquals(chunk, sink.toByteArray());
+        assertEquals(2, frames.size());
+        assertArrayEquals(TESTFR_ACT, frames.get(0));
+        assertArrayEquals(STARTDT_ACT, frames.get(1));
+    }
+
+    @Test
+    public void testFlushDoesNotDuplicate() throws IOException {
+        ByteArrayOutputStream sink = new ByteArrayOutputStream();
+        List<byte[]> frames = new ArrayList<>();
+        MonitoringOutputStream out = new MonitoringOutputStream(sink, frames::add);
+
+        out.write(TESTFR_ACT, 0, TESTFR_ACT.length);
+        out.flush();
+        out.flush();
+
+        assertEquals(1, frames.size());
+        assertArrayEquals(TESTFR_ACT, sink.toByteArray());
+    }
+
+    @Test
+    public void testSplitWriteReassemblesFrame() throws IOException {
+        ByteArrayOutputStream sink = new ByteArrayOutputStream();
+        List<byte[]> frames = new ArrayList<>();
+        MonitoringOutputStream out = new MonitoringOutputStream(sink, frames::add);
+
+        out.write(TESTFR_ACT, 0, 3); // 0x68,0x04,0x43
+        assertTrue(frames.isEmpty());
+        out.write(TESTFR_ACT, 3, 3); // 0x00,0x00,0x00
+
+        assertArrayEquals(TESTFR_ACT, sink.toByteArray());
+        assertEquals(1, frames.size());
+        assertArrayEquals(TESTFR_ACT, frames.get(0));
+    }
+
+    private static byte[] concat(byte[] a, byte[] b) {
+        byte[] out = new byte[a.length + b.length];
+        System.arraycopy(a, 0, out, 0, a.length);
+        System.arraycopy(b, 0, out, a.length, b.length);
+        return out;
+    }
+}

gateway/forecast-south/src/test/java/cn/oleauto/south/MonitoringSocketTest.java

@@ -0,0 +1,128 @@
+package cn.oleauto.south;
+
+import org.junit.Test;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * {@link MonitoringSocketFactory} / {@link MonitoringSocket} loopback 集成测试：
+ * 工厂产物类型、流缓存、收发回调与字节透传端到端验证。
+ */
+public class MonitoringSocketTest {
+
+    private static final byte[] TESTFR_ACT = {0x68, 0x04, 0x43, 0x00, 0x00, 0x00};
+
+    @Test
+    public void testFactoryCreateSocketReturnsMonitoringSocket() throws Exception {
+        MonitoringSocketFactory factory = new MonitoringSocketFactory(null, null, null);
+        try (Socket s = factory.createSocket()) {
+            assertTrue(s instanceof MonitoringSocket);
+            assertFalse(s.isConnected());
+        }
+    }
+
+    @Test
+    public void testStreamsAreMonitoredAndCached() throws Exception {
+        try (ServerSocket server = new ServerSocket(0)) {
+            MonitoringSocketFactory factory = new MonitoringSocketFactory(null, null, null);
+            try (MonitoringSocket socket = (MonitoringSocket) factory.createSocket()) {
+                socket.connect(new InetSocketAddress("127.0.0.1", server.getLocalPort()));
+                assertTrue(socket.isConnected());
+
+                OutputStream os1 = socket.getOutputStream();
+                OutputStream os2 = socket.getOutputStream();
+                assertTrue(os1 instanceof MonitoringOutputStream);
+                assertSame(os1, os2);
+
+                InputStream is1 = socket.getInputStream();
+                InputStream is2 = socket.getInputStream();
+                assertTrue(is1 instanceof MonitoringInputStream);
+                assertSame(is1, is2);
+            }
+        }
+    }
+
+    @Test
+    public void testEndToEndSentFrameCallbackAndBytesDelivered() throws Exception {
+        try (ServerSocket server = new ServerSocket(0)) {
+            List<byte[]> sentFrames = new ArrayList<>();
+            MonitoringSocketFactory factory = new MonitoringSocketFactory(null, sentFrames::add, null);
+            try (MonitoringSocket socket = (MonitoringSocket) factory.createSocket()) {
+                socket.connect(new InetSocketAddress("127.0.0.1", server.getLocalPort()));
+                socket.getOutputStream().write(TESTFR_ACT, 0, TESTFR_ACT.length);
+                socket.getOutputStream().flush();
+
+                try (Socket accepted = server.accept()) {
+                    byte[] received = readFully(accepted.getInputStream(), TESTFR_ACT.length);
+                    assertArrayEquals(TESTFR_ACT, received);
+                }
+                // 发送切帧回调在 write 时同步触发
+                assertEquals(1, sentFrames.size());
+                assertArrayEquals(TESTFR_ACT, sentFrames.get(0));
+            }
+        }
+    }
+
+    @Test
+    public void testEndToEndReceivedFrameCallback() throws Exception {
+        try (ServerSocket server = new ServerSocket(0)) {
+            List<byte[]> receivedFrames = new ArrayList<>();
+            AtomicInteger closed = new AtomicInteger();
+            MonitoringSocketFactory factory = new MonitoringSocketFactory(
+                    receivedFrames::add, null, closed::incrementAndGet);
+            try (MonitoringSocket socket = (MonitoringSocket) factory.createSocket()) {
+                socket.connect(new InetSocketAddress("127.0.0.1", server.getLocalPort()));
+                try (Socket accepted = server.accept()) {
+                    accepted.getOutputStream().write(TESTFR_ACT, 0, TESTFR_ACT.length);
+                    accepted.getOutputStream().flush();
+
+                    // 客户端读取，触发切帧回调
+                    InputStream is = socket.getInputStream();
+                    byte[] buf = new byte[TESTFR_ACT.length];
+                    int total = 0;
+                    long deadline = System.currentTimeMillis() + 5000;
+                    while (total < buf.length && System.currentTimeMillis() < deadline) {
+                        int n = is.read(buf, total, buf.length - total);
+                        if (n == -1) {
+                            break;
+                        }
+                        total += n;
+                    }
+                    assertEquals(buf.length, total);
+                    assertArrayEquals(TESTFR_ACT, buf);
+                    // 切帧回调在读时同步触发
+                    assertEquals(1, receivedFrames.size());
+                    assertArrayEquals(TESTFR_ACT, receivedFrames.get(0));
+                }
+            }
+        }
+    }
+
+    private static byte[] readFully(InputStream in, int n) throws Exception {
+        byte[] buf = new byte[n];
+        int total = 0;
+        long deadline = System.currentTimeMillis() + 5000;
+        while (total < n && System.currentTimeMillis() < deadline) {
+            int r = in.read(buf, total, n - total);
+            if (r == -1) {
+                break;
+            }
+            total += r;
+        }
+        assertEquals(n, total);
+        return buf;
+    }
+}