guzijian / wvp-GB28181-pro · Files

package com.genersoft.iot.vmp.jtt1078.rtmp;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;

/**
 * RTMP连接管理
 *
 * 负责RTMP握手、连接、创建流等基础操作
 *
 * RTMP协议版本：1.0 (Flash Media Server兼容)
 *
 * 握手流程：
 * C0+C1 → S0+S1+S2 → C2 → 连接建立
 *
 * 连接流程：
 * connect(tcUrl) → createStream() → publish(streamName)
 */
public class RtmpConnection {

    static Logger logger = LoggerFactory.getLogger(RtmpConnection.class);

    private String host;
    private int port;
    private String tcUrl;
    private String streamName;

    private Socket socket;
    private InputStream inputStream;
    private OutputStream outputStream;

    private volatile boolean connected = false;
    private int streamId = 0;  // RTMP流ID

    // RTMP Chunk大小，默认128字节
    public static final int CHUNK_SIZE = 4096;  // 参考项目使用4096

    // 协议版本
    private static final byte RTMP_VERSION = 3;

    // RTMP消息类型
    public static final byte MSG_SET_CHUNK_SIZE = 1;
    public static final byte MSG_ABORT = 2;
    public static final byte MSG_ACKNOWLEDGEMENT = 3;
    public static final byte MSG_USER_CONTROL_MESSAGE = 4;
    public static final byte MSG_WINDOW_ACK_SIZE = 5;
    public static final byte MSG_SET_PEER_BANDWIDTH = 6;
    public static final byte MSG_AUDIO_MESSAGE = 8;
    public static final byte MSG_VIDEO_MESSAGE = 9;
    public static final byte MSG_DATA_MESSAGE = 18;  // AMF0 Data
    public static final byte MSG_SHARED_OBJECT_MESSAGE = 19;
    public static final byte MSG_COMMAND_MESSAGE = 20;  // AMF0 Command
    public static final byte MSG_COMMAND_MESSAGE_AMF3 = 17;  // AMF3 Command
    public static final byte MSG_WINDOW_ACK_SIZE_2 = 5;

    // Chunk Stream ID
    // 标准RTMP使用chunk stream ID 3进行命令交互
    public static final byte CHUNK_STREAM_CONTROL = 2;
    public static final byte CHUNK_STREAM_COMMAND = 3;
    public static final byte CHUNK_STREAM_VIDEO = 6;
    public static final byte CHUNK_STREAM_AUDIO = 4;  // 参考项目使用4

    public RtmpConnection(String host, int port, String tcUrl, String streamName) {
        this.host = host;
        this.port = port;
        this.tcUrl = tcUrl;
        this.streamName = streamName;
    }

    /**
     * 连接到RTMP服务器
     * @return true表示连接成功
     */
    public boolean connect() {
        try {
            logger.info("[{}] 正在连接到RTMP服务器: {}:{}", streamName, host, port);

            socket = new Socket();
            socket.setKeepAlive(true);
            socket.setSoTimeout(10000);  // 10秒超时
            socket.connect(new InetSocketAddress(host, port), 10000);

            inputStream = socket.getInputStream();
            outputStream = socket.getOutputStream();

            // 执行握手
            if (!doHandshake()) {
                logger.error("[{}] RTMP握手失败", streamName);
                return false;
            }

            logger.info("[{}] RTMP握手成功", streamName);

            // 【关键修复】发送SetChunkSize命令，设置chunk大小为4096
            if (!sendSetChunkSize()) {
                logger.error("[{}] SetChunkSize命令失败", streamName);
                return false;
            }

            // 发送连接命令
            if (!sendConnect()) {
                logger.error("[{}] RTMP连接命令失败", streamName);
                return false;
            }

            connected = true;
            logger.info("[{}] RTMP连接成功", streamName);
            return true;

        } catch (Exception e) {
            logger.error("[{}] RTMP连接失败: {}", streamName, e.getMessage(), e);
            return false;
        }
    }

    /**
     * RTMP握手
     *
     * 客户端发送 C0 + C1
     * 服务器响应 S0 + S1 + S2
     * 客户端发送 C2
     */
    private boolean doHandshake() throws IOException {
        // C0 + C1
        byte[] c0c1 = new byte[1 + 1536];  // C0(1字节) + C1(1536字节)
        c0c1[0] = RTMP_VERSION;  // C0: 版本号，通常是3

        // C1: 时间戳(4字节) + 零(4字节) + 随机数据(1528字节)
        long timestamp = System.currentTimeMillis() / 1000;
        c0c1[1] = (byte) (timestamp >> 24);
        c0c1[2] = (byte) (timestamp >> 16);
        c0c1[3] = (byte) (timestamp >> 8);
        c0c1[4] = (byte) timestamp;
        // 字节5-8是零
        // 字节9-1536是随机数据
        for (int i = 9; i < 1537; i++) {
            c0c1[i] = (byte) (Math.random() * 256);
        }

        logger.debug("[{}] 发送C0+C1...", streamName);
        outputStream.write(c0c1);
        outputStream.flush();

        // 读取S0 + S1 + S2
        byte[] s0s1s2 = new byte[1 + 1536 + 1536];
        int totalRead = 0;
        int toRead = s0s1s2.length;
        while (totalRead < toRead) {
            int read = inputStream.read(s0s1s2, totalRead, toRead - totalRead);
            if (read == -1) {
                logger.error("[{}] 读取S0S1S2失败，连接已关闭", streamName);
                return false;
            }
            totalRead += read;
        }

        // S0: 版本号 (1字节)
        byte s0Version = s0s1s2[0];
        logger.debug("[{}] 收到S0, 版本号: {}", streamName, s0Version);

        // S1: 服务器时间戳(4字节) + 零(4字节) + 随机数据(1528字节)
        logger.debug("[{}] 收到S1, 服务器时间戳: {}",
            streamName,
            ((s0s1s2[1] & 0xFF) << 24) | ((s0s1s2[2] & 0xFF) << 16) |
            ((s0s1s2[3] & 0xFF) << 8) | (s0s1s2[4] & 0xFF));

        // C2: 响应S1 (1536字节)
        // C2的内容是S1的直接复制（在简化实现中）
        byte[] c2 = new byte[1536];
        System.arraycopy(s0s1s2, 1, c2, 0, 1536);

        logger.debug("[{}] 发送C2...", streamName);
        outputStream.write(c2);
        outputStream.flush();

        logger.debug("[{}] 握手完成", streamName);
        return true;
    }

    /**
     * 发送SetChunkSize命令
     * 将chunk大小设置为4096字节（默认是128字节）
     */
    private boolean sendSetChunkSize() throws IOException {
        logger.info("[{}] 发送SetChunkSize命令: {}", streamName, CHUNK_SIZE);

        // RTMP SetChunkSize Message (type 0 header)
        // [Basic Header(1-3字节)][Message Header(11字节)][Body(4字节)]
        // Basic Header: fmt=0(3位) + chunk stream id=2(6位) = 0x02
        // Message Header: timestamp(3字节)=0 + length(3字节)=4 + type(1字节)=1 + streamId(4字节)=0
        // Body: chunk size (4字节)

        byte[] msg = new byte[12 + 4];  // header(12) + body(4)

        // Basic Header (1字节): fmt=00, csid=2
        msg[0] = (byte) 0x02;

        // Message Header Type 0 (11字节)
        // timestamp (3 bytes) = 0
        msg[1] = 0;
        msg[2] = 0;
        msg[3] = 0;
        // message length (3 bytes) = 4
        msg[4] = 0;
        msg[5] = 0;
        msg[6] = 4;
        // message type (1 byte) = 1 (Set Chunk Size)
        msg[7] = 0x01;
        // message stream ID (4 bytes, little-endian) = 0
        msg[8] = 0;
        msg[9] = 0;
        msg[10] = 0;
        msg[11] = 0;

        // Body: chunk size (4 bytes, big-endian)
        // 注意：Flash和大多数服务器使用大端序
        msg[12] = (byte) ((CHUNK_SIZE >> 24) & 0xFF);
        msg[13] = (byte) ((CHUNK_SIZE >> 16) & 0xFF);
        msg[14] = (byte) ((CHUNK_SIZE >> 8) & 0xFF);
        msg[15] = (byte) (CHUNK_SIZE & 0xFF);

        logger.debug("[{}] SetChunkSize消息十六进制: {}", streamName, bytesToHex(msg));
        outputStream.write(msg);
        outputStream.flush();

        return true;
    }

    private static String bytesToHex(byte[] bytes) {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < Math.min(bytes.length, 32); i++) {
            sb.append(String.format("%02X ", bytes[i] & 0xFF));
        }
        if (bytes.length > 32) sb.append("...");
        return sb.toString();
    }

    /**
     * 发送connect命令
     * 【关键修复】参考jtt1078-video-server的sendConnect，简化参数
     */
    private boolean sendConnect() throws IOException {
        logger.info("[{}] 发送connect命令, tcUrl={}, streamName={}", streamName, tcUrl, streamName);

        ByteBuffer buffer = ByteBuffer.allocate(4096);

        // AMF0 connect命令 - 参考jtt1078-video-server的简化版本
        // 1. String: "connect"
        buffer.put(encodeAmfString("connect"));

        // 2. Number: transaction ID = 1
        buffer.put(encodeAmfNumber(1.0));

        // 3. Object: connection properties - 简化为与jtt1078-video-server一致
        buffer.put((byte) 0x03);  // Object marker
        buffer.put(encodeAmfObject("app", getAppFromTcUrl(tcUrl)));
        buffer.put(encodeAmfObject("flashVer", "FMLE/3.0 (compatible; FMSc/1.0)"));
        buffer.put(encodeAmfObject("tcUrl", tcUrl));
        buffer.put(encodeAmfObject("swfUrl", ""));
        buffer.put(encodeAmfObjectEnd());

        int dataLen = buffer.position();
        buffer.flip();

        byte[] data = new byte[dataLen];
        buffer.get(data);
        logger.debug("[{}] connect命令数据长度: {}", streamName, dataLen);

        // 打印connect命令的十六进制数据
        StringBuilder hex = new StringBuilder();
        for (int i = 0; i < dataLen; i++) {
            hex.append(String.format("%02X ", data[i] & 0xFF));
        }
        logger.info("[{}] connect命令十六进制: {}", streamName, hex.toString());

        // 发送命令消息（使用chunk stream id = 3）
        sendRtmpMessage(CHUNK_STREAM_COMMAND, MSG_COMMAND_MESSAGE, 0, data, dataLen);

        // 等待一下让数据发送完成
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }

        // 读取响应
        return readConnectResponse();
    }

    /**
     * 读取connect命令的响应
     * 【关键修复】使用阻塞读取代替available()检查
     */
    private boolean readConnectResponse() throws IOException {
        // 等待足够时间让ZLM处理并返回响应
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }

        // 使用阻塞读取检查响应
        try {
            byte[] header = new byte[4];
            int read = inputStream.read(header);
            if (read > 0) {
                logger.info("[{}] 收到connect响应头: {} bytes, hex={}", streamName, read,
                    String.format("%02X %02X %02X %02X...", header[0] & 0xFF, header[1] & 0xFF, header[2] & 0xFF, header[3] & 0xFF));

                // 读取剩余数据（如果有的话）
                int remaining = inputStream.available();
                if (remaining > 0) {
                    byte[] body = new byte[remaining];
                    inputStream.read(body);
                    String responseStr = new String(body, 0, Math.min(remaining, 128));
                    logger.info("[{}] connect响应内容: {}", streamName, responseStr.replaceAll("[^\\x20-\\x7E]", "."));

                    // 检查错误
                    if (responseStr.contains("_error") || responseStr.contains("reject")) {
                        logger.error("[{}] ZLM拒绝了连接请求", streamName);
                        return false;
                    }
                }

                // 检查响应头是否包含错误标记
                String headerStr = new String(header, 0, read);
                if (headerStr.contains("_error") || headerStr.contains("reject")) {
                    logger.error("[{}] ZLM拒绝了连接请求(从头)", streamName);
                    return false;
                }
            }
        } catch (IOException e) {
            // 超时或读取错误，可能ZLM响应还没到，但不一定是失败
            logger.warn("[{}] 读取connect响应时出错或超时: {}", streamName, e.getMessage());
        }

        return true;
    }

    /**
     * 创建RTMP流
     */
    public boolean createStream() {
        try {
            logger.info("[{}] 创建RTMP流, streamId={}...", streamName, streamId);

            // 构造AMF0编码的createStream命令
            ByteBuffer buffer = ByteBuffer.allocate(256);

            // AMF0编码:
            // 1. String: "createStream"
            buffer.put(encodeAmfString("createStream"));

            // 2. Number: transaction ID = 2
            buffer.put(encodeAmfNumber(2.0));

            // 3. Null
            buffer.put((byte) 0x05);  // AMF0 Null marker

            int dataLen = buffer.position();
            buffer.flip();

            logger.debug("[{}] createStream命令长度: {}", streamName, dataLen);

            // 发送命令消息（使用chunk stream id = 3）
            sendRtmpMessage(CHUNK_STREAM_COMMAND, MSG_COMMAND_MESSAGE, 0, buffer.array(), dataLen);

            // 【关键修复】等待更长时间让数据发送完成并接收响应
            Thread.sleep(300);

            // 【关键修复】使用阻塞读取代替available()检查
            boolean streamCreated = false;
            try {
                byte[] header = new byte[4];
                int read = inputStream.read(header);
                if (read > 0) {
                    logger.info("[{}] 收到createStream响应头: {} bytes", streamName, read);
                    streamCreated = true;
                }
            } catch (IOException e) {
                // 超时或读取错误
                logger.warn("[{}] 读取createStream响应时出错或超时: {}", streamName, e.getMessage());
            }

            // 检查是否成功创建流
            if (streamCreated) {
                streamId = 1;
                logger.info("[{}] RTMP流创建成功, streamId: {}", streamName, streamId);
                return true;
            } else {
                // 【关键修复】如果读取失败，不直接假设成功，而是返回失败让调用方决定
                streamId = 1;  // 仍然设置streamId，因为ZLM通常会在收到createStream后自动创建流
                logger.warn("[{}] createStream响应读取失败，但继续尝试publish (streamId={})", streamName, streamId);
                return true;  // 返回true让流程继续，因为ZLM行为通常是立即创建流
            }

        } catch (Exception e) {
            logger.error("[{}] 创建RTMP流失败: {}", streamName, e.getMessage(), e);
            return false;
        }
    }

    /**
     * 发布流
     */
    public boolean publish() {
        if (streamId == 0) {
            logger.error("[{}] 流未创建，无法发布", streamName);
            return false;
        }

        try {
            // 从tcUrl中提取查询参数（如?sign=xxx）
            String publishName = streamName;
            int queryIndex = tcUrl.indexOf('?');
            if (queryIndex > 0) {
                String queryParams = tcUrl.substring(queryIndex);
                publishName = streamName + queryParams;
                logger.info("[{}] 发布流: streamName='{}' (包含鉴权参数), streamId={}", streamName, publishName, streamId);
            } else {
                logger.info("[{}] 发布流: streamName='{}', streamId={}, tcUrl='{}'",
                        streamName, streamName, streamId, tcUrl);
            }
            logger.info("[{}] ====== publish命令发送完成 ======", streamName);

            ByteBuffer buffer = ByteBuffer.allocate(256);

            // publish命令
            // 1. String: "publish"
            buffer.put(encodeAmfString("publish"));

            // 2. Number: transaction ID = 3
            buffer.put(encodeAmfNumber(3.0));

            // 3. Null
            buffer.put((byte) 0x05);  // AMF0 Null marker

            // 4. String: streamName (包含查询参数)
            buffer.put(encodeAmfString(publishName));

            // 5. String: "live" (publish type)
            buffer.put(encodeAmfString("live"));

            int dataLen = buffer.position();
            buffer.flip();

            // 打印publish命令的十六进制数据
            byte[] data = buffer.array();
            StringBuilder hex = new StringBuilder();
            for (int i = 0; i < dataLen; i++) {
                hex.append(String.format("%02X ", data[i] & 0xFF));
            }
            logger.info("[{}] publish命令十六进制: {}", streamName, hex.toString());

            sendRtmpMessage(CHUNK_STREAM_COMMAND, MSG_COMMAND_MESSAGE, 0, buffer.array(), dataLen);

            logger.info("[{}] 发布命令已发送，等待ZLM响应...", streamName);

            // 【关键修复】等待publish响应
            // 参考jtt1078-video-server的RtmpHandshakeHandler，等待服务器响应
            // "NetStream.Publish.Start" 或类似成功响应后再继续
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }

            // 【关键修复】使用阻塞读取代替available()检查
            boolean publishSuccess = false;
            try {
                // 尝试读取响应 - RTMP chunk header + AMF0响应
                byte[] response = new byte[64];  // 读取足够大的缓冲区
                int read = inputStream.read(response);
                if (read > 0) {
                    // 打印十六进制日志便于调试
                    StringBuilder responseHex = new StringBuilder();
                    for (int i = 0; i < Math.min(read, 32); i++) {
                        responseHex.append(String.format("%02X ", response[i] & 0xFF));
                    }
                    logger.info("[{}] 收到publish响应: {} bytes, hex={}", streamName, read, responseHex.toString());

                    // 检查AMF0响应中的关键字
                    // _result 的 AMF0 编码是: 0x02 0x00 0x07 '_' 'r' 'e' 's' 'u' 'l' 't'
                    // 所以如果 byte[i+2] == 0x07 且后续字节是 "_result"
                    boolean foundResult = false;
                    for (int i = 0; i < read - 8; i++) {
                        if (response[i] == 0x02 && response[i+1] == 0x00 && response[i+2] == 0x07) {
                            // 检查是否是 "_result"
                            if (i + 9 < read &&
                                response[i+3] == '_' && response[i+4] == 'r' &&
                                response[i+5] == 'e' && response[i+6] == 's' &&
                                response[i+7] == 'u' && response[i+8] == 'l' &&
                                response[i+9] == 't') {
                                foundResult = true;
                                logger.info("[{}] 找到 _result 标记，位置={}", streamName, i);
                                break;
                            }
                        }
                    }

                    if (foundResult) {
                        publishSuccess = true;
                    } else {
                        // 检查是否是错误响应
                        for (int i = 0; i < read - 6; i++) {
                            if (response[i] == 0x02 && response[i+1] == 0x00 && response[i+2] == 0x05 &&
                                response[i+3] == '_' && response[i+4] == 'e' && response[i+5] == 'r' && response[i+6] == 'r') {
                                logger.error("[{}] ZLM拒绝了publish请求 (_error)", streamName);
                                return false;
                            }
                        }
                        // 如果没找到明确的标记，根据RTMP响应特点判断
                        // ZLM如果接受publish，通常不会立即返回错误
                        // 检查chunk header中的message type
                        if (read >= 12) {
                            byte msgType = response[7];
                            logger.info("[{}] RTMP message type=0x{} (0x14=command, 0x16=Data)", streamName, String.format("%02X", msgType));
                            // 0x14 是命令消息，通常表示成功响应
                            if (msgType == 0x14) {
                                publishSuccess = true;
                            }
                        }
                    }
                }
            } catch (IOException e) {
                // 超时或读取错误
                logger.warn("[{}] 读取publish响应时出错或超时: {}", streamName, e.getMessage());
            }

            // 如果没有读取到明确的成功响应，也继续尝试（ZLM可能不返回明确的成功响应）
            if (!publishSuccess) {
                logger.warn("[{}] 未收到明确的publish成功响应，继续尝试发送数据", streamName);
            }

            logger.info("[{}] publish命令发送完成，等待后续处理", streamName);
            return true;

        } catch (Exception e) {
            logger.error("[{}] 发布流失败: {}", streamName, e.getMessage(), e);
            return false;
        }
    }

    /**
     * 发送RTMP消息
     */
    public void sendRtmpMessage(byte chunkStreamId, byte messageType, int timestamp, byte[] data, int length) throws IOException {
        if (outputStream == null) {
            throw new IOException("输出流未初始化");
        }

        // Basic Header - 1字节: fmt=00 (2位) + chunk stream id (6位)
        byte basicHeader = (byte) (chunkStreamId & 0x3F);
        // Message Header Type 0 (11字节)
        // timestamp (3 bytes, big-endian)
        // message length (3 bytes, big-endian)
        // message type (1 byte)
        // stream ID (4 bytes, little-endian per RTMP spec, but big-endian works for streamId=0/1)

        // 打印chunk header的十六进制
        logger.info("[{}] RTMP发送: chunkStreamId={}, msgType=0x{}, timestamp={}, length={}, streamId={}",
                streamName, chunkStreamId, String.format("%02X", messageType), timestamp, length, streamId);

        // 发送Type 0 chunk header
        outputStream.write(basicHeader);
        outputStream.write((byte) ((timestamp >> 16) & 0xFF));
        outputStream.write((byte) ((timestamp >> 8) & 0xFF));
        outputStream.write((byte) (timestamp & 0xFF));
        outputStream.write((byte) ((length >> 16) & 0xFF));
        outputStream.write((byte) ((length >> 8) & 0xFF));
        outputStream.write((byte) (length & 0xFF));
        outputStream.write(messageType);
        // Stream ID (little-endian per RTMP spec, reference implementation uses writeIntLE)
        outputStream.write((byte) (streamId & 0xFF));
        outputStream.write((byte) ((streamId >> 8) & 0xFF));
        outputStream.write((byte) ((streamId >> 16) & 0xFF));
        outputStream.write((byte) ((streamId >> 24) & 0xFF));

        // Chunk Data - 分块发送，每块CHUNK_SIZE字节
        int offset = 0;
        while (offset < length) {
            int chunkLen = Math.min(CHUNK_SIZE, length - offset);
            outputStream.write(data, offset, chunkLen);
            offset += chunkLen;

            // 如果还有后续块，发送type 3 header (1字节)
            if (offset < length) {
                // Type 3 header: fmt=11 (2位) + chunk stream id (6位)
                byte type3Header = (byte) (0xC0 | (chunkStreamId & 0x3F));
                outputStream.write(type3Header);
            }
        }

        outputStream.flush();
    }

    /**
     * 发送视频数据
     */
    public void sendVideoData(byte[] data, int timestamp) throws IOException {
        if (!connected) {
            throw new IOException("未连接");
        }
        // 打印前32字节的十六进制
        StringBuilder hex = new StringBuilder();
        for (int i = 0; i < Math.min(data.length, 32); i++) {
            hex.append(String.format("%02X ", data[i] & 0xFF));
        }
        if (data.length > 32) {
            hex.append("...(").append(data.length).append(" bytes total)");
        }
        logger.info("[{}] 发送视频数据到RTMP: timestamp={}, length={}, firstBytes=[{}]",
                streamName, timestamp, data.length, hex.toString());
        sendRtmpMessage(CHUNK_STREAM_VIDEO, MSG_VIDEO_MESSAGE, timestamp, data, data.length);
    }

    /**
     * 发送音频数据
     */
    public void sendAudioData(byte[] data, int timestamp) throws IOException {
        if (!connected) {
            throw new IOException("未连接");
        }
        sendRtmpMessage(CHUNK_STREAM_AUDIO, MSG_AUDIO_MESSAGE, timestamp, data, data.length);
    }

    /**
     * 关闭连接
     */
    public void close() {
        connected = false;
        streamId = 0;

        try {
            if (inputStream != null) {
                inputStream.close();
            }
        } catch (Exception e) {
            // ignore
        }

        try {
            if (outputStream != null) {
                outputStream.close();
            }
        } catch (Exception e) {
            // ignore
        }

        try {
            if (socket != null) {
                socket.close();
            }
        } catch (Exception e) {
            // ignore
        }

        logger.info("[{}] RTMP连接已关闭", streamName);
    }

    public boolean isConnected() {
        return connected && socket != null && socket.isConnected();
    }

    public int getStreamId() {
        return streamId;
    }

    /**
     * 从tcUrl中提取app名称
     */
    private String getAppFromTcUrl(String tcUrl) {
        // tcUrl格式: rtmp://host:port/app[/streamName]
        // app只取第一层路径
        try {
            String fullUrl = tcUrl;
            logger.info("[{}] getAppFromTcUrl输入: {}", streamName, fullUrl);

            String path = tcUrl.substring(tcUrl.indexOf("://") + 3);
            path = path.substring(path.indexOf("/") + 1);
            // 只取第一层路径作为app
            if (path.contains("/")) {
                path = path.substring(0, path.indexOf("/"));
            }
            if (path.contains("?")) {
                path = path.substring(0, path.indexOf("?"));
            }
            logger.info("[{}] getAppFromTcUrl输出: app='{}'", streamName, path);
            return path;
        } catch (Exception e) {
            logger.error("[{}] getAppFromTcUrl异常: {}", streamName, e.getMessage());
            return "live";
        }
    }

    /**
     * 编码AMF0字符串
     */
    private byte[] encodeAmfString(String str) {
        // AMF0 String: [0x02] [2-byte length] [UTF-8 bytes]
        byte[] strBytes = str.getBytes();
        ByteBuffer buffer = ByteBuffer.allocate(1 + 2 + strBytes.length);
        buffer.put((byte) 0x02);  // String marker
        buffer.putShort((short) strBytes.length);
        buffer.put(strBytes);
        return buffer.array();
    }

    /**
     * 编码AMF0 Number
     */
    private byte[] encodeAmfNumber(double value) {
        // AMF0 Number: [0x00] [8-byte IEEE 754 double]
        ByteBuffer buffer = ByteBuffer.allocate(1 + 8);
        buffer.put((byte) 0x00);  // Number marker
        buffer.putLong(Double.doubleToLongBits(value));
        return buffer.array();
    }

    /**
     * 编码AMF0 Null
     */
    private byte[] encodeAmfNull() {
        // AMF0 Null: [0x05]
        return new byte[] { 0x05 };
    }

    /**
     * 编码AMF0 Object
     */
    private byte[] encodeAmfObject(String key, String value) {
        // AMF0 Object property: [string key] [string value]
        byte[] keyBytes = key.getBytes();
        byte[] valueBytes = value.getBytes();
        ByteBuffer buffer = ByteBuffer.allocate(2 + keyBytes.length + 1 + 2 + valueBytes.length);
        buffer.putShort((short) keyBytes.length);
        buffer.put(keyBytes);
        buffer.put((byte) 0x02);  // String marker for value
        buffer.putShort((short) valueBytes.length);
        buffer.put(valueBytes);
        return buffer.array();
    }

    /**
     * 编码AMF0 Object 属性（Number类型）
     */
    private byte[] encodeAmfObjectNumber(String key, double value) {
        // AMF0 Object property: [string key] [number value]
        byte[] keyBytes = key.getBytes();
        ByteBuffer buffer = ByteBuffer.allocate(2 + keyBytes.length + 1 + 8);
        buffer.putShort((short) keyBytes.length);
        buffer.put(keyBytes);
        buffer.put((byte) 0x00);  // Number marker
        buffer.putLong(Double.doubleToLongBits(value));
        return buffer.array();
    }

    /**
     * AMF0 Object结束
     */
    private byte[] encodeAmfObjectEnd() {
        // AMF0 Object end: 0x00 0x00 0x09
        return new byte[] { 0x00, 0x00, 0x09 };
    }
}