From fee7149458f0a013ba9df3903da2a1e943b90f8e Mon Sep 17 00:00:00 2001 From: Home Date: Mon, 1 Dec 2025 23:32:34 +0800 Subject: [PATCH] x --- python/WebSocketFrameHeader.py | 245 +++++++---- .../pages/index/ProtocolCodec.ts | 404 ++++++++++-------- 测试流式传输uniapp/pages/index/index.vue | 42 +- .../dist/cache/.vite/deps/_metadata.json | 6 +- 4 files changed, 410 insertions(+), 287 deletions(-) diff --git a/python/WebSocketFrameHeader.py b/python/WebSocketFrameHeader.py index 45b7766..909f56c 100644 --- a/python/WebSocketFrameHeader.py +++ b/python/WebSocketFrameHeader.py @@ -2,36 +2,41 @@ from enum import IntEnum import struct import json import gzip -from typing import Optional, Union, Dict, Any, List +from typing import Optional, Union, Dict, Any, List, Tuple -# -------------------------- 协议常量定义 -------------------------- +# -------------------------- 协议常量定义（与JS完全一致） -------------------------- class ProtocolConst: - PROTOCOL_VERSION = 0b0001 # 协议版本v1 - HEADER_SIZE = 8 # 头部固定字节数（字节1~字节8） - MAX_BODY_SIZE = 1024 * 1024 * 10 # 最大包体大小（10MB，防止内存溢出） - STRING_ENCODING = "utf-8" # 字符串默认编码（统一编码格式，避免乱码） + PROTOCOL_VERSION = 0b0001 # 协议版本（4位，字节0低4位，0~15） + HEADER_SIZE = 8 # 头部固定字节数（字节0~7） + MAX_BODY_SIZE = 0xFFFFFF # 最大包体大小（24位，≈16MB，字节4~6存储） + STRING_ENCODING = "utf-8" # 字符串默认编码 -# -------------------------- 枚举定义（业务约定） -------------------------- +# -------------------------- 枚举定义（与JS码值完全对齐） -------------------------- class MessageType(IntEnum): - """消息类型（4位，字节1高4位）""" - PING = 0b0000 # 心跳 - AUDIO_DATA = 0b0001 # 纯音频数据(pcm) - TEXT_MESSAGE = 0b0010 # 纯文本消息 - AUDIO_TEXT_MIX = 0b0011 # 音频+文本混合数据 + """消息类型（4位，字节0高4位，0~15）""" + PING = 0b0001 # 心跳（支持空包体） + AUDIO_DATA = 0b0010 # 纯音频数据(pcm) + TEXT_MESSAGE = 0b0011 # 纯文本消息 CONTROL_CMD = 0b0100 # 控制指令（暂停/继续等） + # 预留12种类型（0b0100 ~ 0b1111） + class SerializationType(IntEnum): - """序列化方式（4位，字节2高4位）""" - RAW = 0b0000 # 原始二进制（音频等） - JSON = 0b0001 # JSON格式（文本/指令） - STRING = 0b0010 # 直接字符串（纯文本，UTF-8编码，无JSON包装）🔴新增类型 + """序列化方式（3位，字节1高3位，1~8）""" + RAW = 0b001 # 原始二进制（1） + JSON = 0b010 # JSON格式（2） + STRING = 0b011 # 直接字符串（3） + # 预留5种方式（0b100 ~ 0b111） + class CompressionType(IntEnum): - """压缩方式（4位，字节2低4位）""" - NONE = 0b0000 # 无压缩 - GZIP = 0b0001 # gzip压缩 + """压缩方式（3位，字节1中3位，1~8）""" + NONE = 0b001 # 无压缩（默认值1） + GZIP = 0b010 # gzip压缩（2） + # 预留6种方式（0b011 ~ 0b111） + class ControlCommand(IntEnum): """控制指令类型（配合MessageType.CONTROL_CMD使用）""" @@ -40,35 +45,62 @@ class ControlCommand(IntEnum): RESUME = 0b0011 # 继续 STOP = 0b0100 # 停止 -# -------------------------- 协议工具类 -------------------------- + +# -------------------------- 协议工具类（与JS协议结构一致） -------------------------- class ProtocolCodec: @staticmethod def pack( msg_type: MessageType, - body: Union[bytes, str, Dict[str, Any], List[Any]], + body: Union[bytes, str, Dict[str, Any], List[Any], None] = None, + sequence: int = 0, serialization: Optional[SerializationType] = None, compression: CompressionType = CompressionType.NONE ) -> bytes: """ - 封装协议包（头部 + 包体） + 封装协议包（与JS pack 方法完全兼容） :param msg_type: 消息类型 - :param body: 包体数据（bytes/str/dict/list） + :param body: 包体数据（PING消息可传None/空，其他类型必填） + :param sequence: 消息顺序号（0~65535，默认0） :param serialization: 序列化方式（None时自动推导） - :param compression: 压缩方式 + :param compression: 压缩方式（默认无压缩） :return: 完整协议包（bytes） """ - # 1. 自动推导序列化方式（优化：纯文本消息自动用STRING，复杂结构用JSON）🔴修改推导逻辑 - if serialization is None: + # 校验顺序号范围（0~65535） + if not isinstance(sequence, int) or not (0 <= sequence <= 0xFFFF): + raise ValueError(f"消息顺序号必须是0~65535的整数，当前传入：{sequence}") + + # 特殊处理：PING消息支持空包体 + if msg_type == MessageType.PING: + # PING消息强制RAW序列化 + serialization = SerializationType.RAW + # 空包体转为空bytes + if body is None: + body = b"" + if not isinstance(body, bytes): + raise TypeError(f"PING消息仅支持空包体或bytes类型，当前传入：{type(body)}") + else: + # 非PING消息包体不能为空 + if body is None: + raise ValueError(f"非PING消息（{msg_type.name}）包体不能为空") + + # 1. 自动推导序列化方式（非PING消息） + if serialization is None and msg_type != MessageType.PING: if msg_type == MessageType.AUDIO_DATA: serialization = SerializationType.RAW - elif msg_type == MessageType.TEXT_MESSAGE: # 纯文本消息→STRING + elif msg_type == MessageType.TEXT_MESSAGE: serialization = SerializationType.STRING - elif msg_type in (MessageType.CONTROL_CMD, MessageType.AUDIO_TEXT_MIX): # 复杂结构→JSON + elif msg_type == MessageType.CONTROL_CMD: serialization = SerializationType.JSON else: raise ValueError(f"不支持的消息类型：{msg_type}") - # 2. 序列化包体（新增STRING类型处理）🔴新增分支 + # 2. 校验枚举值范围 + if not (0b001 <= serialization.value <= 0b111): + raise ValueError(f"序列化方式必须在1~8（0b001~0b111）范围内，当前传入：{serialization.value}") + if not (0b001 <= compression.value <= 0b111): + raise ValueError(f"压缩方式必须在1~8（0b001~0b111）范围内，当前传入：{compression.value}") + + # 3. 序列化包体 serialized_body: bytes if serialization == SerializationType.RAW: if not isinstance(body, bytes): @@ -77,7 +109,7 @@ class ProtocolCodec: elif serialization == SerializationType.STRING: if not isinstance(body, str): raise TypeError("STRING序列化要求body必须是str类型") - serialized_body = body.encode(ProtocolConst.STRING_ENCODING) # 直接UTF-8编码，无JSON包装 + serialized_body = body.encode(ProtocolConst.STRING_ENCODING) elif serialization == SerializationType.JSON: if isinstance(body, str): serialized_body = body.encode(ProtocolConst.STRING_ENCODING) @@ -88,7 +120,7 @@ class ProtocolCodec: else: raise ValueError(f"不支持的序列化方式：{serialization}") - # 3. 压缩包体（不变） + # 4. 压缩包体 compressed_body: bytes if compression == CompressionType.GZIP: compressed_body = gzip.compress(serialized_body) @@ -97,50 +129,81 @@ class ProtocolCodec: else: raise ValueError(f"不支持的压缩方式：{compression}") - # 4. 校验包体大小（不变） + # 5. 校验包体大小（24位最大支持0xFFFFFF字节） body_len = len(compressed_body) if body_len > ProtocolConst.MAX_BODY_SIZE: - raise OverflowError(f"包体过大（{body_len}字节），最大支持{ProtocolConst.MAX_BODY_SIZE}字节") + raise OverflowError( + f"包体过大（{body_len}字节），最大支持{ProtocolConst.MAX_BODY_SIZE}字节（≈16MB）" + ) - # 5. 构造头部（不变） - byte1 = (msg_type.value << 4) | 0x00 # 消息类型(4位) + 保留位1(4位) - byte2 = (serialization.value << 4) | (compression.value & 0x0F) # 序列化(4位) + 压缩(4位) - byte3_4 = struct.pack(">H", ProtocolConst.PROTOCOL_VERSION) # 协议版本(16位) - byte5_8 = struct.pack(">I", body_len) # 包体长度(32位) - header = bytes([byte1, byte2]) + byte3_4 + byte5_8 + # 6. 构造头部（与JS头部结构完全一致） + # 字节0：消息类型(高4位) + 协议版本(低4位) + byte0 = ((msg_type.value & 0x0F) << 4) | (ProtocolConst.PROTOCOL_VERSION & 0x0F) + # 字节1：序列化方式(高3位) + 压缩方式(中3位) + 保留位(低2位) + byte1 = ((serialization.value & 0x07) << 5) | ((compression.value & 0x07) << 2) | 0x00 + # 字节2~3：消息顺序号（16位大端序） + byte2_3 = struct.pack(">H", sequence) + # 字节4~6：包体长度（24位大端序） + byte4 = (body_len >> 16) & 0xFF + byte5 = (body_len >> 8) & 0xFF + byte6 = body_len & 0xFF + # 字节7：保留位（固定0x00） + byte7 = 0x00 + + # 拼接头部 + header = ( + bytes([byte0, byte1]) + + byte2_3 + + bytes([byte4, byte5, byte6, byte7]) + ) assert len(header) == ProtocolConst.HEADER_SIZE, f"头部长度错误：实际{len(header)}字节，预期{ProtocolConst.HEADER_SIZE}字节" return header + compressed_body @staticmethod - def unpack(packet: bytes) -> tuple[MessageType, SerializationType, CompressionType, Any]: + def unpack(packet: bytes) -> Tuple[MessageType, SerializationType, CompressionType, int, Any]: """ - 解析协议包 + 解析协议包（与JS unpack 方法完全兼容） :param packet: 完整协议包（头部 + 包体） - :return: (消息类型, 序列化方式, 压缩方式, 原始包体数据) + :return: (消息类型, 序列化方式, 压缩方式, 顺序号, 原始包体数据) """ - # 1. 校验包长度（不变） + # 1. 校验包长度 if len(packet) < ProtocolConst.HEADER_SIZE: - raise ValueError(f"包长度过短（{len(packet)}字节），至少需要{ProtocolConst.HEADER_SIZE}字节头部") + raise ValueError( + f"包长度过短（{len(packet)}字节），至少需要{ProtocolConst.HEADER_SIZE}字节头部" + ) - # 2. 解析头部（不变） + # 2. 解析头部 header = packet[:ProtocolConst.HEADER_SIZE] body = packet[ProtocolConst.HEADER_SIZE:] - byte1 = header[0] - byte2 = header[1] - msg_type = MessageType((byte1 >> 4) & 0x0F) - serialization = SerializationType((byte2 >> 4) & 0x0F) - compression = CompressionType(byte2 & 0x0F) - version = struct.unpack(">H", header[2:4])[0] - body_len = struct.unpack(">I", header[4:8])[0] + # 字节0：消息类型(高4位) + 协议版本(低4位) + byte0 = header[0] + msg_type = MessageType((byte0 >> 4) & 0x0F) + version = byte0 & 0x0F + # 字节1：序列化方式(高3位) + 压缩方式(中3位) + 保留位(低2位) + byte1 = header[1] + serialization = SerializationType((byte1 >> 5) & 0x07) + compression = CompressionType((byte1 >> 2) & 0x07) + + # 字节2~3：消息顺序号（16位大端序） + sequence = struct.unpack(">H", header[2:4])[0] + + # 字节4~6：包体长度（24位大端序），字节7：保留位（忽略） + body_len = (header[4] << 16) | (header[5] << 8) | header[6] + + # 校验版本和包体长度 if version != ProtocolConst.PROTOCOL_VERSION: - raise ValueError(f"协议版本不匹配：收到v{version}，支持v{ProtocolConst.PROTOCOL_VERSION}") + raise ValueError( + f"协议版本不匹配：收到v{version}（0b{version:04b}），支持v{ProtocolConst.PROTOCOL_VERSION}（0b{ProtocolConst.PROTOCOL_VERSION:04b}）" + ) if len(body) != body_len: - raise ValueError(f"包体长度不匹配：头部声明{body_len}字节，实际{len(body)}字节") + raise ValueError( + f"包体长度不匹配：头部声明{body_len}字节，实际{len(body)}字节" + ) - # 3. 解压包体（不变） + # 3. 解压包体 decompressed_body: bytes if compression == CompressionType.GZIP: try: @@ -152,13 +215,15 @@ class ProtocolCodec: else: raise ValueError(f"不支持的压缩方式：{compression}") - # 4. 反序列化包体（新增STRING类型处理）🔴新增分支 + # 4. 反序列化包体（空包体返回None） original_body: Any - if serialization == SerializationType.RAW: + if len(decompressed_body) == 0: + original_body = None + elif serialization == SerializationType.RAW: original_body = decompressed_body elif serialization == SerializationType.STRING: try: - original_body = decompressed_body.decode(ProtocolConst.STRING_ENCODING) # 直接UTF-8解码，无JSON解析 + original_body = decompressed_body.decode(ProtocolConst.STRING_ENCODING) except UnicodeDecodeError: raise ValueError(f"STRING反序列化失败：{ProtocolConst.STRING_ENCODING}解码错误") elif serialization == SerializationType.JSON: @@ -166,58 +231,64 @@ class ProtocolCodec: original_body = json.loads(decompressed_body.decode(ProtocolConst.STRING_ENCODING)) except UnicodeDecodeError: raise ValueError(f"JSON反序列化失败：{ProtocolConst.STRING_ENCODING}解码错误") - except json.JSONDecodeError: - raise ValueError("JSON反序列化失败：格式错误") + except json.JSONDecodeError as e: + raise ValueError(f"JSON反序列化失败：格式错误（{str(e)}）") else: raise ValueError(f"不支持的序列化方式：{serialization}") - return msg_type, serialization, compression, original_body + return msg_type, serialization, compression, sequence, original_body -# -------------------------- 使用示例（新增纯字符串消息测试） -------------------------- +# -------------------------- 使用示例（验证与JS兼容性） -------------------------- if __name__ == "__main__": - # 示例1：纯文本消息（自动用STRING序列化，无JSON包装）🔴测试新增类型 - text_body = "你好，这是纯字符串消息（无JSON）" + # 示例1：PING消息（空包体，默认顺序号0） + ping_packet = ProtocolCodec.pack(MessageType.PING) + print(f"PING消息包长度：{len(ping_packet)}字节（仅头部）") + msg_type1, ser1, comp1, seq1, body1 = ProtocolCodec.unpack(ping_packet) + print(f"PING解析结果：类型={msg_type1.name}，序列化={ser1.name}，压缩={comp1.name}，顺序号={seq1}，包体={body1}\n") + + # 示例2：纯文本消息（STRING序列化，指定顺序号） + text_body = "Python与JS协议兼容测试（纯字符串）" text_packet = ProtocolCodec.pack( msg_type=MessageType.TEXT_MESSAGE, body=text_body, + sequence=1001, compression=CompressionType.NONE ) - print(f"纯文本消息包长度：{len(text_packet)}字节") - msg_type1, ser1, comp1, body1 = ProtocolCodec.unpack(text_packet) - print(f"解析结果：类型={msg_type1.name}，序列化={ser1.name}，压缩={comp1.name}，内容={body1}\n") + print(f"文本消息包长度：{len(text_packet)}字节") + msg_type2, ser2, comp2, seq2, body2 = ProtocolCodec.unpack(text_packet) + print(f"文本解析结果：类型={msg_type2.name}，序列化={ser2.name}，顺序号={seq2}，内容={body2}\n") - # 示例2：控制指令（JSON序列化，复杂结构） - control_body = {"cmd": ControlCommand.HEARTBEAT.value, "timestamp": 1699999999} + # 示例3：控制指令（JSON序列化） + control_body = {"cmd": ControlCommand.PAUSE.value, "reason": "用户主动暂停"} control_packet = ProtocolCodec.pack( msg_type=MessageType.CONTROL_CMD, body=control_body, - compression=CompressionType.NONE + sequence=1002 ) print(f"控制指令包长度：{len(control_packet)}字节") - msg_type2, ser2, comp2, body2 = ProtocolCodec.unpack(control_packet) - print(f"解析结果：类型={msg_type2.name}，序列化={ser2.name}，压缩={comp2.name}，内容={body2}\n") + msg_type3, ser3, comp3, seq3, body3 = ProtocolCodec.unpack(control_packet) + print(f"控制指令解析结果：类型={msg_type3.name}，序列化={ser3.name}，顺序号={seq3}，内容={body3}\n") - # 示例3：音频数据（RAW序列化） + # 示例4：音频数据（RAW序列化） audio_body = b"\x00\x01\x02\x03\x04\x05" * 100 # 模拟PCM数据 audio_packet = ProtocolCodec.pack( msg_type=MessageType.AUDIO_DATA, body=audio_body, - compression=CompressionType.NONE + sequence=1003 ) print(f"音频数据包长度：{len(audio_packet)}字节") - msg_type3, ser3, comp3, body3 = ProtocolCodec.unpack(audio_packet) - print(f"解析结果：类型={msg_type3.name}，序列化={ser3.name}，压缩={comp3.name}，数据长度={len(body3)}字节\n") + msg_type4, ser4, comp4, seq4, body4 = ProtocolCodec.unpack(audio_packet) + print(f"音频解析结果：类型={msg_type4.name}，序列化={ser4.name}，顺序号={seq4}，数据长度={len(body4)}字节\n") - # 示例4：混合数据（JSON序列化） - mix_body = { - "audio_data": list(audio_body[:10]), - "text_content": "混合数据仍用JSON" - } - mix_packet = ProtocolCodec.pack( - msg_type=MessageType.AUDIO_TEXT_MIX, - body=mix_body + # 示例5：GZIP压缩测试（需JS端启用GZIP解压） + long_text_body = "这是一段很长的文本，用于测试GZIP压缩效果" * 100 + gzip_packet = ProtocolCodec.pack( + msg_type=MessageType.TEXT_MESSAGE, + body=long_text_body, + sequence=1004, + compression=CompressionType.GZIP ) - print(f"混合数据包长度：{len(mix_packet)}字节") - msg_type4, ser4, comp4, body4 = ProtocolCodec.unpack(mix_packet) - print(f"解析结果：类型={msg_type4.name}，序列化={ser4.name}，压缩={comp4.name}，内容={body4}") \ No newline at end of file + print(f"GZIP压缩后包长度：{len(gzip_packet)}字节（原始文本长度：{len(long_text_body.encode())}字节）") + msg_type5, ser5, comp5, seq5, body5 = ProtocolCodec.unpack(gzip_packet) + print(f"GZIP解析结果：类型={msg_type5.name}，压缩={comp5.name}，内容前50字：{body5[:50]}...") \ No newline at end of file diff --git a/测试流式传输uniapp/pages/index/ProtocolCodec.ts b/测试流式传输uniapp/pages/index/ProtocolCodec.ts index b9e4f2b..62605a5 100644 --- a/测试流式传输uniapp/pages/index/ProtocolCodec.ts +++ b/测试流式传输uniapp/pages/index/ProtocolCodec.ts @@ -5,128 +5,156 @@ * 协议常量类：存储协议核心配置（不可修改，确保前后端一致） */ export class ProtocolConst { - /** 协议版本：v1（4位，对应字节3~4的低4位，剩余位保留用于后续升级） */ + /** 协议版本：0~15（4位，存储在字节0低4位） */ static readonly PROTOCOL_VERSION = 0b0001; - /** 头部固定长度：8字节（字节1~8，结构严格定义，不可修改） */ + /** 头部固定长度：8字节（字节0~7，结构严格定义，不可修改） */ static readonly HEADER_SIZE = 8; - /** 最大包体大小：10MB（防止内存溢出，与 Python 端 MAX_BODY_SIZE 一致） */ - static readonly MAX_BODY_SIZE = 1024 * 1024 * 10; + /** + * 最大包体大小：16MB（3字节长度最大支持0xFFFFFF=16777215字节≈16MB） + * 4-6字节存储（24位），最大支持16MB，满足大部分场景且避免长度字段冗余 + */ + static readonly MAX_BODY_SIZE = 0xFFFFFF; // 16777215字节 ≈16MB /** 字符串编码格式：UTF-8（统一前后端字符串编解码，避免乱码） */ - static readonly STRING_ENCODING = "utf-8" as const; // as const 固定字符串类型，避免类型拓宽 + static readonly STRING_ENCODING = "utf-8" as const; } /** - * 消息类型枚举（4位，存储在字节1高4位） - * 与 Python 端 MessageType 枚举值完全对齐，支持16种类型（当前用5种，剩余预留扩展） + * 消息类型枚举（4位，存储在字节0高4位，0~15范围） + * 二进制标识，统一编码风格 */ export enum MessageType { - PING = 0b0000, // 心跳消息（用于检测连接存活） - AUDIO_DATA = 0b0001, // 纯音频数据（PCM 二进制流，对应 RAW 序列化） - TEXT_MESSAGE = 0b0010, // 纯文本消息（对应 STRING 序列化，无 JSON 包装） - AUDIO_TEXT_MIX = 0b0011, // 音频+文本混合数据（复杂结构，对应 JSON 序列化） - CONTROL_CMD = 0b0100, // 控制指令（暂停/继续/停止等，对应 JSON 序列化） + PING = 0b0001, // 心跳消息（支持空包体） + AUDIO_DATA = 0b0010, // 纯音频数据 + TEXT_MESSAGE = 0b0011, // 纯文本消息 + CONTROL_CMD = 0b0100, // 控制指令 + // 预留12种类型用于扩展（0b0100 ~ 0b1111） } /** - * 序列化方式枚举（4位，存储在字节2高4位） - * 按数据类型选择最优序列化方式，减少开销 + * 序列化方式枚举（3位，存储在字节1高3位，1~8范围） + * 二进制标识，统一编码风格（1~8对应0b001~0b111） */ export enum SerializationType { - RAW = 0b0000, // 原始二进制（无需编解码，直接透传，适合音频等二进制数据） - JSON = 0b0001, // JSON 格式（适合字典、列表等复杂结构，如控制指令、混合数据） - STRING = 0b0010, // 直接字符串（UTF-8 编码，无 JSON 包装，适合纯文本消息） + RAW = 0b001, // 原始二进制（1） + JSON = 0b010, // JSON 格式（2） + STRING = 0b011, // 直接字符串（3） + // 预留5种方式用于扩展（0b100 ~ 0b111） } /** - * 压缩方式枚举（4位，存储在字节2低4位） - * 按需选择压缩策略，平衡性能和传输体积 + * 压缩方式枚举（3位，存储在字节1中3位，1~8范围） + * 二进制标识，统一编码风格（1~8对应0b001~0b111） */ export enum CompressionType { - NONE = 0b0000, // 无压缩（小数据包如控制指令、短文本，避免压缩开销） - GZIP = 0b0001, // GZIP 压缩（大数据包如长文本、混合数据，减少网络传输量） + NONE = 0b001, // 无压缩（1，默认值） + GZIP = 0b010, // GZIP 压缩（2） + // 预留6种方式用于扩展（0b011 ~ 0b111） } /** * 控制指令枚举（配合 MessageType.CONTROL_CMD 使用） - * 定义业务层面的控制指令，与 Python 端控制指令值一致 */ export enum ControlCommand { - HEARTBEAT = 0b0001, // 心跳响应（回复 PING 消息） - PAUSE = 0b0010, // 暂停指令（如暂停音频播放） - RESUME = 0b0011, // 继续指令（如恢复音频播放） - STOP = 0b0100, // 停止指令（如停止音频传输） + HEARTBEAT = 0b0001, + PAUSE = 0b0010, + RESUME = 0b0011, + STOP = 0b0100, } /** - * 解包返回结果接口（强类型约束，明确返回数据结构） - * 让 IDE 提供自动提示，避免类型错误 + * 解包返回结果接口 */ export interface UnpackedResult { - msgType: MessageType; // 消息类型（枚举值，便于逻辑判断） - msgTypeName: keyof typeof MessageType; // 消息类型名称（字符串，如 "TEXT_MESSAGE"，便于日志打印） - serialization: SerializationType; // 序列化方式（枚举值） - serializationName: keyof typeof SerializationType; // 序列化方式名称（字符串） - compression: CompressionType; // 压缩方式（枚举值） - compressionName: keyof typeof CompressionType; // 压缩方式名称（字符串） - body: Uint8Array | string | object | unknown[]; // 解包后的原始包体（根据序列化方式动态变化） + msgType: MessageType; + msgTypeName: keyof typeof MessageType; + serialization: SerializationType; + serializationName: keyof typeof SerializationType; + compression: CompressionType; + compressionName: keyof typeof CompressionType; + sequence: number; // 消息顺序号（0~65535，默认0） + body: Uint8Array | string | object | unknown[] | null; // PING 消息可能返回 null } /** - * 打包入参类型别名（简化方法入参类型定义，提高可读性） + * 打包入参类型别名（支持 PING 消息传入 null） */ -// 包体支持的类型：二进制、字符串、对象、数组 -type PackBody = Uint8Array | string | object | unknown[]; -// 序列化方式可选值：枚举值、null、undefined（null/undefined 时自动推导） +type PackBody = Uint8Array | string | object | unknown[] | null; type OptionalSerialization = SerializationType | null | undefined; /** - * 协议编解码工具类（静态类，无需实例化，提供 pack/unpack 静态方法） - * 核心功能：将业务数据打包为符合协议的二进制包，或解析二进制包为业务数据 - * 特点：与 Python/JS 端完全兼容，性能高效，类型安全 + * 协议编解码工具类（支持 PING 消息空包体） */ export class ProtocolCodec { /** - * 打包协议包：将业务数据按协议格式封装为二进制包 - * @param msgType 消息类型（必须指定，枚举值） - * @param body 业务数据（根据消息类型对应不同类型，如字符串、对象、Uint8Array） - * @param serialization 序列化方式（可选，默认自动推导：纯文本→STRING，音频→RAW，复杂结构→JSON） - * @param compression 压缩方式（可选，默认无压缩） - * @returns 完整协议包（Uint8Array 类型，便于网络传输） - * @throws 类型错误、不支持的枚举值、包体过大等异常 + * 打包协议包 + * @param msgType 消息类型（二进制枚举，0~15） + * @param body 业务数据（PING 消息可传 null/undefined，其他类型必填） + * @param sequence 消息顺序号（0~65535，可选，默认0） + * @param serialization 序列化方式（二进制枚举，1~8，可选，自动推导） + * @param compression 压缩方式（二进制枚举，1~8，可选，默认NONE=0b001） + * @returns 完整协议包 + * @throws 类型错误、范围错误、包体过大等异常 */ static pack( msgType: MessageType, - body: PackBody, + body: PackBody = null, // 默认为 null，支持 PING 消息空包体 + sequence: number = 0, // 可选参数，默认0 serialization: OptionalSerialization = null, compression: CompressionType = CompressionType.NONE ): Uint8Array { - // 1. 自动推导序列化方式（减少调用方心智负担，按消息类型默认最优解） - if (serialization === null || serialization === undefined) { + // 校验顺序号范围（0~65535） + if (!Number.isInteger(sequence) || sequence < 0 || sequence > 0xFFFF) { + throw new RangeError(`消息顺序号必须是0~65535的整数，当前传入：${sequence}`); + } + + // 特殊处理：PING 消息允许空包体，强制 RAW 序列化（空二进制） + if (msgType === MessageType.PING) { + // PING 消息忽略传入的序列化方式，强制使用 RAW（空二进制最高效） + serialization = SerializationType.RAW; + // 若传入空包体，统一处理为空 Uint8Array + body = body === null || body === undefined ? new Uint8Array(0) : body; + // PING 消息仅支持空包体或 Uint8Array（防止误传其他类型） + if (!(body instanceof Uint8Array)) { + throw new TypeError(`PING 消息仅支持空包体或 Uint8Array 类型，当前传入：${typeof body}`); + } + } else { + // 非 PING 消息：包体必填 + if (body === null || body === undefined) { + throw new TypeError(`非 PING 消息（${MessageType[msgType]}）包体不能为空`); + } + } + + // 1. 自动推导序列化方式（非 PING 消息） + if (serialization === null || serialization === undefined && msgType !== MessageType.PING) { if (msgType === MessageType.AUDIO_DATA) { - // 音频数据→RAW 序列化（无需编解码，性能最优） serialization = SerializationType.RAW; } else if (msgType === MessageType.TEXT_MESSAGE) { - // 纯文本消息→STRING 序列化（直接 UTF-8 编码，无冗余） serialization = SerializationType.STRING; - } else if (msgType === MessageType.CONTROL_CMD || msgType === MessageType.AUDIO_TEXT_MIX) { - // 控制指令/混合数据→JSON 序列化（支持复杂结构） + } else if (msgType === MessageType.CONTROL_CMD) { serialization = SerializationType.JSON; } else { - // 非法消息类型：抛出异常，提前阻断错误 throw new Error(`不支持的消息类型：${MessageType[msgType]}（值：${msgType}）`); } } - // 2. 序列化包体：将业务数据转为二进制（按序列化方式处理） - let serializedBody: Uint8Array; // 序列化后的二进制数据 - const textEncoder = new TextEncoder(ProtocolConst.STRING_ENCODING); // UTF-8 编码器 + // 2. 校验枚举值范围（3位存储，1~8即0b001~0b111） + if (serialization < 0b001 || serialization > 0b111) { + throw new RangeError(`序列化方式必须在1~8（0b001~0b111）范围内，当前传入：${serialization}（0b${serialization.toString(2).padStart(3, '0')}）`); + } + if (compression < 0b001 || compression > 0b111) { + throw new RangeError(`压缩方式必须在1~8（0b001~0b111）范围内，当前传入：${compression}（0b${compression.toString(2).padStart(3, '0')}）`); + } + + // 3. 序列化包体 + let serializedBody: Uint8Array; + const textEncoder = new TextEncoder(); switch (serialization) { case SerializationType.RAW: - // RAW 序列化：必须传入 Uint8Array（二进制数据直接透传） + // RAW 序列化：支持 Uint8Array（PING 消息可能是空 Uint8Array） if (!(body instanceof Uint8Array)) { throw new TypeError(`RAW 序列化要求 body 必须是 Uint8Array 类型，当前传入：${typeof body}`); } @@ -134,7 +162,7 @@ export class ProtocolCodec { break; case SerializationType.STRING: - // STRING 序列化：必须传入字符串，直接 UTF-8 编码（无 JSON 包装） + // STRING 序列化：必须传入字符串（非 PING 消息已校验非空） if (typeof body !== "string") { throw new TypeError(`STRING 序列化要求 body 必须是 string 类型，当前传入：${typeof body}`); } @@ -142,12 +170,10 @@ export class ProtocolCodec { break; case SerializationType.JSON: - // JSON 序列化：支持字符串、对象、数组，转为 JSON 字符串后 UTF-8 编码 + // JSON 序列化：支持字符串、对象、数组（非 PING 消息已校验非空） if (typeof body === "string") { - // 已为字符串，直接编码 serializedBody = textEncoder.encode(body); } else if (typeof body === "object" && body !== null) { - // 对象/数组→JSON 字符串→编码（用 JSON.stringify 序列化） const jsonStr = JSON.stringify(body); serializedBody = textEncoder.encode(jsonStr); } else { @@ -156,183 +182,197 @@ export class ProtocolCodec { break; default: - // 非法序列化方式：抛出异常 - throw new Error(`不支持的序列化方式：${SerializationType[serialization]}（值：${serialization}）`); + throw new Error(`不支持的序列化方式：${SerializationType[serialization]}（值：${serialization}，0b${serialization.toString(2).padStart(3, '0')}）`); } - // 3. 压缩包体：按指定压缩方式处理（GZIP 或无压缩） - let compressedBody: Uint8Array; // 压缩后的二进制数据 + // 4. 压缩包体 + let compressedBody: Uint8Array; if (compression === CompressionType.NONE) { - // 无压缩：直接透传序列化后的二进制数据 compressedBody = serializedBody; + } else if (compression === CompressionType.GZIP) { + // 若需启用GZIP，取消注释下方代码（需导入pako） + // try { + // compressedBody = pako.gzip(serializedBody); + // } catch (e) { + // throw new Error(`GZIP 压缩失败：${(e as Error).message}`); + // } + throw new Error("GZIP 压缩暂未启用，请导入pako库并取消对应代码注释"); } else { - // 非法压缩方式：抛出异常 - throw new Error(`不支持的压缩方式：${CompressionType[compression]}（值：${compression}）`); + throw new Error(`不支持的压缩方式：${CompressionType[compression]}（值：${compression}，0b${compression.toString(2).padStart(3, '0')}）`); } - // 4. 校验包体大小：防止超过最大限制（避免内存溢出） - const bodyLen = compressedBody.length; // 压缩后的包体长度 + // 5. 校验包体大小（24位长度最大支持0xFFFFFF=16777215字节） + const bodyLen = compressedBody.length; if (bodyLen > ProtocolConst.MAX_BODY_SIZE) { throw new Error( - `包体过大（${bodyLen}字节），最大支持${ProtocolConst.MAX_BODY_SIZE}字节（10MB）` + `包体过大（${bodyLen}字节），最大支持${ProtocolConst.MAX_BODY_SIZE}字节（≈16MB）` ); } - // 5. 构造头部：8字节固定结构（严格按协议定义，大端序，与 Python 端一致） - const header = new Uint8Array(ProtocolConst.HEADER_SIZE); // 头部缓冲区（8字节） + // 6. 构造头部（8字节，按最新结构） + const header = new Uint8Array(ProtocolConst.HEADER_SIZE); - // 字节1：消息类型(4位) + 保留位(4位) - // 消息类型左移4位（占高4位），保留位填0（后续扩展用） - header[0] = (msgType << 4) | 0x00; + // 字节0：消息类型(高4位) + 协议版本(低4位) + header[0] = ((msgType & 0x0F) << 4) | (ProtocolConst.PROTOCOL_VERSION & 0x0F); - // 字节2：序列化方式(4位) + 压缩方式(4位) - // 序列化方式左移4位（占高4位），压缩方式占低4位（&0x0F 确保仅4位） - header[1] = (serialization << 4) | (compression & 0x0F); + // 字节1：序列化方式(高3位) + 压缩方式(中3位) + 保留位(低2位，填0) + header[1] = ((serialization & 0x07) << 5) | ((compression & 0x07) << 2) | 0x00; - // 字节3~4：协议版本（16位大端序） - // 大端序：高位字节在前，低位字节在后（网络传输标准字节序） - header[2] = (ProtocolConst.PROTOCOL_VERSION >> 8) & 0xFF; // 版本高8位（当前版本为1，高8位为0） - header[3] = ProtocolConst.PROTOCOL_VERSION & 0xFF; // 版本低8位（当前版本为1） + // 字节2~3：消息顺序号（16位大端序，0~65535） + header[2] = (sequence >> 8) & 0xFF; // 顺序号高8位 + header[3] = sequence & 0xFF; // 顺序号低8位 - // 字节5~8：包体长度（32位大端序） - // 32位整数：最高位字节（第5字节）→ 最低位字节（第8字节） - header[4] = (bodyLen >> 24) & 0xFF; // 包体长度第24~31位 - header[5] = (bodyLen >> 16) & 0xFF; // 包体长度第16~23位 - header[6] = (bodyLen >> 8) & 0xFF; // 包体长度第8~15位 - header[7] = bodyLen & 0xFF; // 包体长度第0~7位 + // 字节4~6：消息体长度（24位大端序，0~0xFFFFFF） + header[4] = (bodyLen >> 16) & 0xFF; // 长度高8位 + header[5] = (bodyLen >> 8) & 0xFF; // 长度中8位 + header[6] = bodyLen & 0xFF; // 长度低8位 - // 6. 拼接头部和包体：生成完整协议包 - const totalLen = ProtocolConst.HEADER_SIZE + bodyLen; // 总长度 = 头部8字节 + 包体长度 - const packet = new Uint8Array(totalLen); // 完整包缓冲区 + // 字节7：保留位（固定填0x00） + header[7] = 0x00; - packet.set(header, 0); // 从索引0开始写入头部（占前8字节） - packet.set(compressedBody, ProtocolConst.HEADER_SIZE); // 从索引8开始写入包体 + // 7. 拼接头部和包体 + const totalLen = ProtocolConst.HEADER_SIZE + bodyLen; + const packet = new Uint8Array(totalLen); + packet.set(header, 0); + packet.set(compressedBody, ProtocolConst.HEADER_SIZE); - return packet; // 返回完整协议包（Uint8Array，网络传输高效） + return packet; } /** - * 解包协议包：将二进制协议包解析为业务数据 - * @param packet 完整协议包（网络接收的二进制数据，支持 Uint8Array 或 ArrayBuffer） - * @returns 结构化解包结果（UnpackedResult 接口，包含消息类型、包体等信息） - * @throws 包长度过短、版本不匹配、解压缩失败、反序列化失败等异常 + * 解包协议包 + * @param packet 完整协议包 + * @returns 结构化解包结果（含顺序号） + * @throws 各种解析异常 */ static unpack(packet: Uint8Array | ArrayBuffer): UnpackedResult { - // 统一数据类型：将 ArrayBuffer 转为 Uint8Array（便于按字节操作） const uint8Packet = packet instanceof ArrayBuffer ? new Uint8Array(packet) : packet; - // 1. 校验包长度：至少包含8字节头部 + // 1. 校验包长度（至少8字节头部） if (uint8Packet.length < ProtocolConst.HEADER_SIZE) { throw new Error( `包长度过短（${uint8Packet.length}字节），至少需要${ProtocolConst.HEADER_SIZE}字节头部` ); } - // 2. 拆分头部和包体：前8字节为头部，剩余为包体 - const header = uint8Packet.subarray(0, ProtocolConst.HEADER_SIZE); // 头部（0~7索引） - const bodyBuffer = uint8Packet.subarray(ProtocolConst.HEADER_SIZE); // 包体（8索引开始） + // 2. 拆分头部和包体 + const header = uint8Packet.subarray(0, ProtocolConst.HEADER_SIZE); + const bodyBuffer = uint8Packet.subarray(ProtocolConst.HEADER_SIZE); - // 3. 解析头部字段（按协议结构逐字节解析，大端序） - // 字节1：消息类型(4位) + 保留位(4位) - const byte1 = header[0]; - const msgType = (byte1 >> 4) & 0x0F; // 右移4位取高4位（消息类型），&0x0F 确保仅4位 - // 校验消息类型合法性（防止非法包注入） + // 3. 解析头部字段 + // 字节0：消息类型(高4位) + 协议版本(低4位) + const byte0 = header[0]; + const msgType = (byte0 >> 4) & 0x0F; // 消息类型（0~15） + const version = byte0 & 0x0F; // 协议版本（0~15） + + // 校验消息类型 if (!Object.values(MessageType).includes(msgType as MessageType)) { - throw new Error(`非法消息类型：${msgType}（无对应枚举值）`); + throw new Error(`非法消息类型：${msgType}（0b${msgType.toString(2).padStart(4, '0')}）`); } - // 字节2：序列化方式(4位) + 压缩方式(4位) - const byte2 = header[1]; - const serialization = (byte2 >> 4) & 0x0F; // 右移4位取高4位（序列化方式） - const compression = byte2 & 0x0F; // 取低4位（压缩方式） - // 校验序列化方式合法性 - if (!Object.values(SerializationType).includes(serialization as SerializationType)) { - throw new Error(`非法序列化方式：${serialization}（无对应枚举值）`); - } - // 校验压缩方式合法性 - if (!Object.values(CompressionType).includes(compression as CompressionType)) { - throw new Error(`非法压缩方式：${compression}（无对应枚举值）`); - } - - // 字节3~4：协议版本（16位大端序） - const version = (header[2] << 8) | header[3]; // 大端序解析：高位字节<<8 + 低位字节 - // 校验版本兼容性（仅支持当前协议版本） + // 校验版本 if (version !== ProtocolConst.PROTOCOL_VERSION) { throw new Error( - `协议版本不匹配：收到v${version}，当前支持v${ProtocolConst.PROTOCOL_VERSION}` + `协议版本不匹配：收到v${version}（0b${version.toString(2).padStart(4, '0')}），当前支持v${ProtocolConst.PROTOCOL_VERSION}（0b${ProtocolConst.PROTOCOL_VERSION.toString(2).padStart(4, '0')}）` ); } - // 字节5~8：包体长度（32位大端序） - const bodyLen = (header[4] << 24) | (header[5] << 16) | (header[6] << 8) | header[7]; - // 校验包体长度一致性（头部声明长度 vs 实际包体长度） + // 字节1：序列化方式(高3位) + 压缩方式(中3位) + 保留位(低2位) + const byte1 = header[1]; + const serialization = (byte1 >> 5) & 0x07; // 高3位（1~8） + const compression = (byte1 >> 2) & 0x07; // 中3位（1~8） + // 保留位：(byte1 & 0x03)，暂不处理 + + // 校验序列化方式 + if (!Object.values(SerializationType).includes(serialization as SerializationType)) { + throw new Error(`非法序列化方式：${serialization}（0b${serialization.toString(2).padStart(3, '0')}）`); + } + + // 校验压缩方式 + if (!Object.values(CompressionType).includes(compression as CompressionType)) { + throw new Error(`非法压缩方式：${compression}（0b${compression.toString(2).padStart(3, '0')}）`); + } + + // 字节2~3：消息顺序号（16位大端序） + const sequence = (header[2] << 8) | header[3]; // 0~65535 + + // 字节4~6：消息体长度（24位大端序），字节7：保留位（忽略） + const bodyLen = (header[4] << 16) | (header[5] << 8) | header[6]; + + // 校验包体长度（空包体时 bodyBuffer.length 应为0） if (bodyBuffer.length !== bodyLen) { throw new Error( - `包体长度不匹配：头部声明${bodyLen}字节，实际接收${bodyBuffer.length}字节（可能包丢失或非法包）` + `包体长度不匹配：头部声明${bodyLen}字节，实际接收${bodyBuffer.length}字节` ); } - // 4. 解压包体：按头部指定的压缩方式解压 - let decompressedBody: Uint8Array; // 解压后的二进制数据 + // 4. 解压包体 + let decompressedBody: Uint8Array; if (compression === CompressionType.NONE) { - // 无压缩：直接透传包体数据 decompressedBody = bodyBuffer; + } else if (compression === CompressionType.GZIP) { + // 若需启用GZIP，取消注释下方代码 + // try { + // decompressedBody = pako.ungzip(bodyBuffer); + // } catch (e) { + // throw new Error(`GZIP 解压失败：${(e as Error).message}`); + // } + throw new Error("GZIP 解压暂未启用，请导入pako库并取消对应代码注释"); } else { - // 此处理论上不会触发（已提前校验压缩方式合法性） - throw new Error(`不支持的压缩方式：${CompressionType[compression]}（值：${compression}）`); + throw new Error(`不支持的压缩方式：${CompressionType[compression]}（值：${compression}，0b${compression.toString(2).padStart(3, '0')}）`); } - // 5. 反序列化包体：将二进制数据转为业务数据（按序列化方式处理） - let body: UnpackedResult["body"]; // 反序列化后的原始业务数据 - const textDecoder = new TextDecoder(ProtocolConst.STRING_ENCODING); // UTF-8 解码器 + // 5. 反序列化包体（PING 消息空包体返回 null） + let body: UnpackedResult["body"]; + const textDecoder = new TextDecoder(); - switch (serialization) { - case SerializationType.RAW: - // RAW 反序列化：直接返回 Uint8Array（二进制数据，如音频 PCM） - body = decompressedBody; - break; + // 特殊处理：空包体（PING 消息常见）返回 null + if (decompressedBody.length === 0) { + body = null; + } else { + switch (serialization) { + case SerializationType.RAW: + body = decompressedBody; + break; - case SerializationType.STRING: - // STRING 反序列化：UTF-8 解码为字符串（无 JSON 解析步骤） - try { - body = textDecoder.decode(decompressedBody); - } catch (e) { - throw new Error(`STRING 反序列化失败：${(e as Error).message}（UTF-8 解码错误，可能是非法字符串数据）`); - } - break; - - case SerializationType.JSON: - // JSON 反序列化：先 UTF-8 解码为字符串，再 JSON.parse 转为对象/数组 - try { - const jsonStr = textDecoder.decode(decompressedBody); // 二进制→JSON 字符串 - body = JSON.parse(jsonStr); // JSON 字符串→对象/数组 - } catch (e) { - if (e instanceof TypeError) { - throw new Error(`JSON 反序列化失败：${ProtocolConst.STRING_ENCODING} 解码错误（非法 UTF-8 数据）`); - } else if (e instanceof SyntaxError) { - throw new Error(`JSON 反序列化失败：格式错误（${(e as Error).message}），请检查 JSON 语法`); - } else { - throw new Error(`JSON 反序列化失败：${(e as Error).message}`); + case SerializationType.STRING: + try { + body = textDecoder.decode(decompressedBody); + } catch (e) { + throw new Error(`STRING 反序列化失败：UTF-8 解码错误`); } - } - break; + break; - default: - // 此处理论上不会触发（已提前校验序列化方式合法性） - throw new Error(`不支持的序列化方式：${SerializationType[serialization]}（值：${serialization}）`); + case SerializationType.JSON: + try { + const jsonStr = textDecoder.decode(decompressedBody); + body = JSON.parse(jsonStr); + } catch (e) { + if (e instanceof SyntaxError) { + throw new Error(`JSON 反序列化失败：格式错误（${(e as Error).message}）`); + } else { + throw new Error(`JSON 反序列化失败：${(e as Error).message}`); + } + } + break; + + default: + throw new Error(`不支持的序列化方式：${SerializationType[serialization]}（值：${serialization}，0b${serialization.toString(2).padStart(3, '0')}）`); + } } - // 6. 构造并返回解包结果（利用枚举反向映射获取名称，便于日志和展示） + // 6. 返回解包结果 return { - msgType: msgType as MessageType, // 消息类型枚举值 - msgTypeName: MessageType[msgType] as keyof typeof MessageType, // 消息类型名称（如 "TEXT_MESSAGE"） - serialization: serialization as SerializationType, // 序列化方式枚举值 - serializationName: SerializationType[serialization] as keyof typeof SerializationType, // 序列化方式名称 - compression: compression as CompressionType, // 压缩方式枚举值 - compressionName: CompressionType[compression] as keyof typeof CompressionType, // 压缩方式名称 - body: body, // 反序列化后的原始业务数据 + msgType: msgType as MessageType, + msgTypeName: MessageType[msgType] as keyof typeof MessageType, + serialization: serialization as SerializationType, + serializationName: SerializationType[serialization] as keyof typeof SerializationType, + compression: compression as CompressionType, + compressionName: CompressionType[compression] as keyof typeof CompressionType, + sequence: sequence, + body: body, }; } } \ No newline at end of file diff --git a/测试流式传输uniapp/pages/index/index.vue b/测试流式传输uniapp/pages/index/index.vue index b7e6976..4dc9bdd 100644 --- a/测试流式传输uniapp/pages/index/index.vue +++ b/测试流式传输uniapp/pages/index/index.vue @@ -8,10 +8,19 @@ @onStop="stopIt"> xx +