第 96 章 WebSocket Stream：接管与事件流

本章导读

第 95 章的帧与消息是状态机（无传输的库层）；本章的 Stream（xwsstream）把它们接上真实传输：xrtWsStreamAttach/AttachTls 接管已经完成升级的 TCP/TLS 调用方引用（含"缓冲里还剩帧数据"的精确衔接），事件按 MessageBegin → MessageData×N → MessageEnd 发布——正文视图只在当前回调有效，Stream 不聚合完整消息、不为每连接预分配固定缓冲，TCP/TLS 的现有块链直接进帧状态机。发送侧提供 copy 与 ref 引用两种路径（引用在真正写完 socket 后才释放——组播场景同一引用挂 N 连接计数释放）；SendLimit 是 WebSocket 与底层传输待发量的统一硬边界，Backpressure/Writable/Drain 给恢复边沿。Ping 自动回复、Close 唯一发送保证——连接对象的全部生命周期。

96.1引入

第 95 章末的练习让你手搭帧循环——你会发现 80% 的代码在处理"传输接线"：升级后的缓冲里可能已经带着帧数据、TLS 帧头可能跨 record、写背压要同时看两层、Ping 要自动回、Close 只能发一次……这些每个实现都要重写的管线，Stream 层收口成 Attach 一次调用 + 四个事件回调。

最值得先懂的是接管衔接：升级应答刚发完，接收缓冲里可能已经有客户端抢发的第一帧（浏览器经常这样）。iPrefix 参数告诉 Stream"已验证但尚未消费的 HTTP 头长度"——接管时先复制协商出的子协议、精确消费头部、同一缓冲中的帧余量直接进帧状态机，一个字节都不浪费、不错位。失败不会部分接管（传输引用完好归还）。

96.2概念

接管：Attach 的精确衔接

TLS 的特有处理：帧头可能跨 TLS record 边界——Stream 通过受 PlainLimit 限制的 ReadMore 路径累积最多 14 字节的完整帧头（最大帧头尺寸），"不因保留一个不完整前缀而停住"——第 87 章 TLS 流与本章帧状态机的衔接细节已经调好。

事件模型：视图即用即弃

MessageData 的正文视图只在当前同步回调内有效——Stream 不聚合完整消息（大消息不占内存）、不拷贝（块链直通）。你的消费逻辑必须"回调内完成"（解析、转发、写盘）；需要保留就自己复制。这与第 87 章 TLS 流的明文消费、第 91 章 Body 的 Data 视图同一纪律——借用视图的世界里，保留是显式动作。

发送三形态与统一背压

• copy（Text/Binary/Send）：负载复制入队——小消息的简单路径。
• ref（TextRef/BinaryTake/SendRef 等）：传 xnetref{数据, 长度, 释放回调, 上下文}——队列排队期间零复制，真正写完 socket 后才调释放回调。组播场景同一引用挂 N 个连接（计数释放）——一处分配、N 处发送。
• 压缩发送（TextCompressed 等）：按需创建 Deflater，permessage-deflate 协商后的路径（第 97 章讲协商）。

背压是统一水位：SendLimit 同时计 WebSocket 待发与底层传输待发——达到上限 XNET_RESULT_AGAIN；Backpressure/Writable/Drain 事件给出恢复边沿（进入背压、可写恢复、排空）。与第 68 章 TCP 写预算、第 87 章 TLS SendLimit 一脉——每层一个水位、语义相同。

控制帧与关闭协议

Ping 自动回复 Pong（应用无感）；Ping 可主动发起。Close 的保证：唯一发送（重复 Close 不会上线）、超时机制（对端不回应答按超时处理）、对端终态快照（CloseInfo：码+原因，供应用读）。关闭码的语义域与写侧校验是第 97 章组合章的内容——本章记住"Close 调用一次，状态机管协议"。

生命周期与自省

Ref/Destroy 引用计数（与其他连接对象一致）；自省族：State/Role/Protocol/Worker/Tcp/TcpRef/Tls/TlsRef/Deflate/Pending/Writable——连接的全部公开事实；Pause/Paused/Resume 流控（暂停接收，配合下游限速）。

96.3示例

第一个完整程序：引用发送——写完才释放

下面的程序来自 examples/websocket/stream_ref/main.c——零拷贝发送路径的标准样本：

#include <stdio.h>
#include <string.h>

#include <xrt.h>





















 */


/* 引用发送完成后归还业务负载。 */
static void releasePayload(
	ptr pContext,
	cbytes pData,
	size_t iSize
)
{
	(void)pContext;
	(void)iSize;
	xrtFree((ptr)pData);
}



/* 在 Connection 所属 Worker 上发送一个可延迟释放的业务负载。 */
static xnetresult sendPayload(
	xwsstream* pConnection,
	xbytesview Payload
)
{
	bytes pOwned;
	xnetref Ref;
	xnetresult Result;

	pOwned = (bytes)xrtMalloc(Payload.Size);
	if ( pOwned == NULL ) {
		return XNET_RESULT_ERROR;
	}
	memcpy(pOwned, Payload.Data, Payload.Size);
	Ref = (xnetref) {
		pOwned,
		Payload.Size,
		releasePayload,
		NULL
	};
	Result = xrtWsStreamBinaryRef(pConnection, &Ref);
	if ( Result != XNET_RESULT_OK ) {
		xrtFree(pOwned);
	}
	return Result;
}



/* 示例函数由已经建立连接的 Worker 回调调用。 */
int main(void)
{
	(void)sendPayload;
	printf("WebSocket Connection reference example is ready\n");
	return 0;
}

刚才发生了什么。 ① 业务负载由 xrtMalloc 分配（模拟来自上游的消息）；xnetref 把 {指针, 长度, releasePayload 回调, 上下文} 打包。② xrtWsStreamBinaryRef(连接, 负载, Ref) 以引用入队——负载在 Worker 队列里排队期间不复制；发送真正完成（写进 socket）后 releasePayload 才被调用、xrtFree 归还。③ 为什么这很重要：高吞吐场景 copy 路径每次发送都是一次分配+复制；ref 路径一次分配、零中转复制。组播升级（头注释指路）：同一引用挂到 N 个连接——内部计数释放，最后一个写完才真正 free。④ Pending/Writable 配合等待排空——统一背压的读法。

配套的 examples/websocket/stream_tour 是 Stream 层全特性巡检（回环双端）：配置校验、双端 Attach、64 字节回显、自动与手动 Pong、暂停恢复、压缩三形态、干净关闭（code=1000）——把它当本章的可执行契约表跑一遍。

第二个完整程序：扩展协商预览

第二个程序来自 examples/websocket/extension/main.c——升级后的扩展参数遍历，为第 97 章铺垫：

#include <stdio.h>

#include <xrt.h>






















 */


/* 展示逐项读取 Sec-WebSocket-Extensions 和扩展参数。 */
int main(void)
{
	xstrview Text = XRT_STR_LITERAL(
		"permessage-deflate; client_max_window_bits, x-trace"
	);
	xwsextension Extension;
	xhttpnext Next;
	size_t iOffset = 0;

	for ( ;; ) {
		xhttpparam Param;
		size_t iParam = 0;

		Next = xrtWsExtensionNext(
			Text,
			&iOffset,
			&Extension
		);
		if ( Next == XHTTP_NEXT_END ) {
			break;
		}
		if ( Next == XHTTP_NEXT_ERROR ) {
			return 1;
		}
		printf(
			"extension=%.*s\n",
			(int)Extension.Name.Size,
			Extension.Name.Data
		);
		while ( xrtWsExtensionParamNext(
			&Extension,
			&iParam,
			&Param
		) == XHTTP_NEXT_ITEM ) {
			printf(
				"  parameter=%.*s\n",
				(int)Param.Name.Size,
				Param.Name.Data
			);
		}
	}
	return 0;
}

刚才发生了什么。 ① Sec-WebSocket-Extensions 的两层结构：外层按 , 切扩展、内层按 ; 切参数——与第 92 章字段族语法同构，但扩展层有自己的迭代器（x-trace 这样的自定义扩展也用 name + 参数表达）。② client_max_window_bits 是 permessage-deflate 的协商参数之一——第 97 章完整讲协商与压缩链路；这里先建立"扩展是升级后连接的属性、协商决定 Stream 行为（压缩路径、RSV 位）"的认知。③ 扩展迭代是纯字段层操作——调试代理、合规检查器不用建连接就能审计协商结果。

96.4契约

• 接管原子性：Attach 失败不部分接管（传输引用完好）；iPrefix 精确消费已验证头部，同缓冲帧余量直通状态机。
• 事件视图：MessageData 正文仅在当前回调有效；不聚合完整消息、无每连接固定缓冲——块链直通帧状态机。
• TLS 跨 record：帧头跨 TLS record 由 ReadMore 累积（≤14 字节、受 PlainLimit 限）——不停不丢。
• 发送三形态：copy（复制入队）/ ref（引用+延迟释放回调）/ compressed（按需 Deflater）；容量与预算语义同全库。
• 统一背压：SendLimit 同计 WS 与传输待发；AGAIN + Backpressure/Writable/Drain 边沿事件。
• 引用发送：xnetref{数据,长度,释放,上下文}；写完 socket 才释放；同引用可挂 N 连接计数释放。
• Ping/Pong：收到 Ping 自动回 Pong；主动 Ping 可发；流控 Pause/Paused/Resume。
• Close 保证：唯一发送、超时机制、对端终态快照（CloseInfo）；Abort 立即中止。
• 自省族：State/Role/Protocol/Worker/Tcp(+Ref)/Tls(+Ref)/Deflate/Pending/Writable。
• 裁剪：WEBSOCKET_STREAM 独立于帧层；_REF/_TLS/_DEFLATE 子特性按需。

常见错误

onMessageData(pStream, Data, ...) {
	queue_push(g_Queue, Data);   /* 存视图：返回即悬空 */
}

onMessageData(pStream, Data, ...) {
	bytes pCopy = (bytes)xrtMalloc(Data.Size);
	memcpy(pCopy, Data.Data, Data.Size);
	queue_push(g_Queue, pCopy);   /* 副本可异步 */
}
/* 更优：同步完成消费（解析/转发），不引入队列 */

xrtWsStreamBinaryRef(pStream, PayloadView, &Ref);
xrtFree(pBuffer);   /* 入队就释放：队列里的引用悬空 */

/* 所有权随引用移交；releasePayload 回调里释放 */
xrtWsStreamBinaryRef(pStream, PayloadView, &Ref);
/* 此后 pBuffer 归 Stream 管——不再碰它 */

if ( my_ws_queue_size() < MY_LIMIT ) {
	send_more();   /* 传输层可能早已超限 */
}

if ( xrtWsStreamSend(pStream, ...) == XNET_RESULT_AGAIN ) {
	/* 等 Writable/Drain 事件——统一水位的恢复边沿 */
	return;
}

编程练习

小结