第 110 章 xws（上）：连接管理与全链巡检

本章导读

卷九第 95–97 章讲了 WebSocket 的协议核心（帧/消息/流）；xws 扩展库在其上补齐工程连接层：xwsconn 接管已经完成握手的 TCP 或 TLS 流——不负责 URL、DNS、HTTP 请求、证书策略或重定向（那些归第 94 章升级与卷八 TLS），只管连接上的消息事件、背压、关闭与自省。本章三块：事件与视图纪律（消息开始/数据/结束/Ping/Pong/Close 事件，数据视图仅回调内有效）；流控与发送契约（pause/resume 接收流控、高/低水位与 AGAIN 结果——继承网络流的全部词汇）；全链巡检（connection_tour 的七行自检覆盖离线构建、实连接对打、九种发送、异步族、自动 Pong、暂停恢复、干净关闭——本章的可执行契约表）。Future/协程桥与关闭协议收尾。

110.1引入

xws 的分工边界值得先画清楚：核心库（卷九）提供帧、消息、握手、扩展协商、permessage-deflate——可组合的协议零件；xws 扩展层（本章起）把这些零件接到真实连接上，并补齐工程设施（连接事件、引用生命周期、Future 等待、连接组、服务端路由）。契约文档明确“HTTP 客户端/服务器适配、路由和框架型便利对象已经退出核心”——扩展层的定位是高级能力的重建区，不能让对象模型反向污染帧与直接收发路径。这个“核心轻、扩展厚”的分层与 xhttp（核心协议在卷九、应用层在卷十）完全同构。

连接层的第一个设计决定是接管而非创建：xwsconn 接手“已完成握手的流”——谁完成握手（第 94 章升级流程、xws 服务端路由（第 113 章））不重要，连接层只面对“已就绪的传输”。这让它可以平等地站在 TCP 与 TLS 之上（第 97 章 TLS 接管路径）。

110.2概念

事件模型与视图纪律

连接事件：消息开始、数据片段、消息结束、Ping、Pong、Close、错误、最终关闭——比第 96 章 Stream 层多了 Ping/Pong 的独立事件（控制帧不只自动回，也可见）。视图纪律与全库一致：回调中的数据视图只在回调期间有效；跨回调保存必须复制或接管拥有型缓冲。角色规则继承核心层：服务端要求客户端帧有掩码、客户端要求服务端帧无掩码——角色在连接建立时固定，不按单帧猜测。

流控：pause/resume 与水位

1接收消息事件流 → 应用消费

2慢消费pause 暂停接收 → 对端受 TCP 背压 → 内存恒定

3恢复resume 继续——不允许靠无限增长的应用队列解决慢消费者

4发送继承网络流高/低水位、最大排队字节与 AGAIN 结果契约

这是“停消费=停接收”纪律（第 62 章骨架经济学）的连接层实现。发送路径的 AGAIN 与水位词汇与第 96 章完全一致——学过 Stream 层这里零新词。

发送所有权三态（发送路径预告）

copy（返回前复制——短控制帧/短消息）、ref（队列释放时调一次 release——静态或共享大块）、take/buffer（移交已分配内存或网络缓冲链）——第 111 章展开 writer 形态，本章先立“未受理时所有权仍归调用方，受理后只由连接释放”的原子规则。

Future 与协程桥

Future 只包装已建立连接上的发送、关闭、drain 和终态等待——不创建隐藏 HTTP 客户端。两条纪律：取消等待≠销毁连接——调用方按具体操作契约决定继续用、正常关闭还是 abort；协程通过通用 Future/Wait 体系等待（第 58 章）——WebSocket 内不维护第二套调度器。connection_future 示例族演示等待形态。

关闭协议与错误域

正常关闭顺序：发送或响应 Close → 停止新数据发送 → 等对端 Close 或截止时间 → 关闭写方向 → 等传输结束。协议破坏、超限、底层失败可直接 abort——但必须释放所有已受理引用恰好一次（引用所有权的终态保证）。错误模型：协议错误用稳定 WebSocket 域错误码并尽可能映射 RFC Close code；网络/TLS/应用回调错误保留各自错误域——不被压成一个布尔原因（第 4 章原因链在协议层的兑现）。

110.3示例

第一个完整程序：七行全链巡检

下面的程序来自 examples/websocket/connection_tour——xhttp 服务器+客户端回环对接后的连接层全特性自检：

extlibs/xws/examples/websocket/connection_tour/main.c源码 ↗









































 */

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

#include <math.h>
static inline double exampleTimerLimit(int64 Timeout)
{
    return Timeout == XRT_WAIT_FOREVER ? INFINITY : xrtTimer() + (double)Timeout / 1000.0;
}
static inline bool exampleTimerExpired(double Limit)
{
    return xrtTimer() >= Limit;
}
static inline int64 exampleTimerRemaining(double Limit)
{
    double Ms;
    if (Limit == INFINITY) return XRT_WAIT_FOREVER;
    Ms = ceil((Limit - xrtTimer()) * 1000.0);
    return Ms <= 0 ? 0 : Ms >= 0x1p63 ? INT64_MAX : (int64)Ms;
}

/* 等待上限（毫秒）。 */
#define TOUR_DEADLINE_MS INT64_C(10000)
#ifndef TOUR_BACKEND
#define TOUR_BACKEND XNET_PORT_AUTO
#endif
#ifndef TOUR_WORKERS
#define TOUR_WORKERS 2
#endif

/* 共享状态：两侧计数与阶段标志。 */
typedef struct tourstate {
	xatomicptr Client;
	xatomicptr Server;
	xatomic32 ClientDone;
	xatomic32 ServerDone;
	xatomic32 Messages;    /* 服务端收到的完整消息数 */
	xatomic32 Echoes;      /* 客户端收到的回显数 */
	xatomic32 Pings;       /* 服务端观察到的 Ping */
	xatomic32 Pongs;       /* 客户端观察到的 Pong */
	xatomic32 Closed;
	xatomic32 CleanClosed;
	xatomic32 Errors;
	xatomic32 PauseNext;
	bool bIntroOk;         /* 由 ClientDone 的 release/acquire 发布 */
	bool bClientCheckOk;
	char Key[XWS_KEY_CAPACITY];
} tourstate;

static xwsconnevents g_Events;

/* Ref/Take 负载释放计数。 */
static xatomic32 g_Released;

static void tourRelease(ptr pContext, cbytes pData, size_t iSize)
{
	(void)pContext;
	(void)iSize;
	xrtFree((ptr)pData);
	xrtAtomic32FetchAdd(&g_Released, 1, XMEMORY_RELEASE);
}

/* 客户端消息收到回显。 */
static void tourClientEnd(xwsconn* pConnection, ptr pData)
{
	(void)pConnection;
	xrtAtomic32FetchAdd(&((tourstate*)pData)->Echoes, 1, XMEMORY_RELEASE);
}

/* 服务端：回显策略——收一条回一条。 */
static void tourServerEnd(xwsconn* pConnection, ptr pData)
{
	tourstate* pState = (tourstate*)pData;

	if ( xrtWsConnText(pConnection, XRT_STR_LITERAL("ok")) != XNET_RESULT_OK ) {
		xrtAtomic32FetchAdd(&pState->Errors, 1, XMEMORY_RELEASE);
	}
	if ( xrtAtomic32Exchange(&pState->PauseNext, 0, XMEMORY_ACQ_REL) ) {
		xrtWsConnPause(pConnection);
	}
	xrtAtomic32FetchAdd(&pState->Messages, 1, XMEMORY_RELEASE);
}

static void tourMessageEnd(xwsconn* pConnection, ptr pData)
{
	if ( xrtWsConnRole(pConnection) == XWS_ROLE_SERVER ) {
		tourServerEnd(pConnection, pData);
	}
	else {
		tourClientEnd(pConnection, pData);
	}
}

static void tourPing(xwsconn* pConnection, xbytesview Payload,
	ptr pData)
{
	tourstate* pState = (tourstate*)pData;

	(void)Payload;
	if ( xrtWsConnRole(pConnection) == XWS_ROLE_SERVER ) {
		xrtAtomic32FetchAdd(&pState->Pings, 1, XMEMORY_RELEASE);
	}
}

static void tourPong(xwsconn* pConnection, xbytesview Payload,
	ptr pData)
{
	tourstate* pState = (tourstate*)pData;

	if ( (xrtWsConnRole(pConnection) == XWS_ROLE_CLIENT) &&
		(Payload.Size == 2u) && (memcmp(Payload.Data, "pk", 2u) == 0) ) {
		xrtAtomic32FetchAdd(&pState->Pongs, 1, XMEMORY_RELEASE);
	}
}

static void tourClose(xwsconn* pConnection, const xwsconnclose* pClose,
	ptr pData)
{
	tourstate* pState = (tourstate*)pData;

	/* 干净关闭：双方都见到 1000 且握手完整。 */
	if ( (pClose != NULL) &&
		((pClose->Flags & XWS_CONN_CLOSE_CLEAN) != 0u) &&
		((pClose->Flags & XWS_CONN_CLOSE_SENT) != 0u) &&
		((pClose->Flags & XWS_CONN_CLOSE_RECEIVED) != 0u) &&
		(pClose->LocalCode == 1000u) ) {
		/* CloseInfo 快照同样可读（Worker 上）。 */
		xwsconnclose Snapshot;

		if ( xrtWsConnCloseInfo(pConnection, &Snapshot) &&
			(Snapshot.RemoteCode == 1000u) ) {
			xrtAtomic32FetchAdd(&pState->CleanClosed, 1, XMEMORY_RELEASE);
		}
	}
	xrtAtomic32FetchAdd(&pState->Closed, 1, XMEMORY_RELEASE);
}

static void tourError(xwsconn* pConnection, const xerror* pError,
	ptr pData)
{
	(void)pConnection;
	fprintf(stderr, "conn-tour: error kind=%d code=%d\n",
		(int)xrtErrorKind(pError), (int)xrtErrorCode(pError));
	xrtAtomic32FetchAdd(&((tourstate*)pData)->Errors, 1, XMEMORY_RELEASE);
}

/* 服务端 Upgrade 完成：接管回调转移的连接引用。 */
static void tourServerDone(xhttpconn* pHttp, xnetresult Result,
	xwsconn* pConnection, const xerror* pError, ptr pData)
{
	tourstate* pState = (tourstate*)pData;

	(void)pHttp;
	(void)pError;
	if ( (Result == XNET_RESULT_OK) && (pConnection != NULL) ) {
		xrtAtomicPtrStore(&pState->Server, pConnection,
			XMEMORY_RELEASE);
	}
	xrtAtomic32Store(&pState->ServerDone,
		(Result == XNET_RESULT_OK && pConnection != NULL) ? 1 : 2,
		XMEMORY_RELEASE);
}

/* 客户端握手完成：自省 + 三族发送 + 关闭。 */
static void tourClientDone(xhttpcall* pCall, xnetresult Result,
	xwsconn* pConnection, xhttpresponse* pResponse,
	const xerror* pError, ptr pData)
{
	tourstate* pState = (tourstate*)pData;
	xwsclientconfig Config;
	xwsclienthandshake Handshake;
	xhttprequest* pActual = NULL;
	const xhttpfield* pKey;
	xnetstream* pTcp;
	xtlsstream* pTls;

	(void)pCall;
	(void)pError;
	if ( (Result != XNET_RESULT_OK) || (pConnection == NULL) ) {
		xrtHttpResponseDestroy(pResponse);
		xrtAtomic32Store(&pState->ClientDone, 2, XMEMORY_RELEASE);
		return;
	}
	
	 * 离线 RequestCreate/Clone 返回的 Key 不属于这次在线握手。 */
	xrtWsClientConfigInit(&Config);
	Config.Protocols = XRT_STR_LITERAL("chat.v1, chat.v2");
	pActual = xrtHttpCallRequestClone(pCall);
	pKey = pActual == NULL ? NULL : xrtHttpRequestHeader(pActual,
		XRT_STR_LITERAL("Sec-WebSocket-Key"));
	if ( (pKey != NULL) &&
		xrtWsClientCheck(pResponse, pKey->Value, &Config, &Handshake) ) {
		pState->bClientCheckOk = true;
	}
	/* 自省族（Worker 上）：角色/传输/Worker/预算/错误。 */
	pTcp = xrtWsConnTcpRef(pConnection);
	pTls = xrtWsConnTlsRef(pConnection);
	if ( (xrtWsConnRole(pConnection) == XWS_ROLE_CLIENT) &&
		(xrtWsConnTcp(pConnection) != NULL) &&
		(pTcp != NULL) &&
		(xrtWsConnTls(pConnection) == NULL) &&
		(pTls == NULL) &&
		(xrtWsConnWorker(pConnection) != NULL) &&
		!xrtWsConnDeflate(pConnection, NULL) &&
		(xrtWsConnWritable(pConnection) > 0u) &&
		!xrtWsConnPaused(pConnection) &&
		(xrtWsConnError(pConnection) == NULL) ) {
		pState->bIntroOk = true;
	}
	xrtNetStreamDestroy(pTcp);
	xrtTlsStreamDestroy(pTls);
	xrtHttpRequestDestroy(pActual);
	xrtHttpResponseDestroy(pResponse);
	xrtAtomicPtrStore(&pState->Client, pConnection,
		XMEMORY_RELEASE);
	xrtAtomic32Store(&pState->ClientDone, 1, XMEMORY_RELEASE);
}

/* 服务端 HTTP 请求处理：解析升级请求并接管。 */
static void tourRequest(xhttpserver* pServer, xhttpconn* pHttp,
	const xhttpserverrequest* pRequest, ptr pData)
{
	tourstate* pState = (tourstate*)pData;
	xwsserverconfig Config;
	xwsserverhandshake Handshake;

	(void)pServer;
	xrtWsServerConfigInit(&Config);
	Config.Protocols = XRT_STR_LITERAL("chat.v1, chat.v2");
	if ( xrtWsServerCheck(pRequest, &Config, &Handshake) ) {
		if ( xrtWsUpgradeAccept(pHttp, &Config,
			&Handshake, NULL, &g_Events, pState,
			tourServerDone, pState) == XNET_RESULT_OK ) return;
	}
	xrtAtomic32FetchAdd(&pState->Errors, 1, XMEMORY_RELEASE);
	xrtAtomic32Store(&pState->ServerDone, 2, XMEMORY_RELEASE);
}

static bool tourWait32Min(xatomic32* pFlag, uint32 iMinimum)
{
	double iEnd = exampleTimerLimit(TOUR_DEADLINE_MS);

	while ( xrtAtomic32Load(pFlag, XMEMORY_ACQUIRE) < iMinimum ) {
		if ( exampleTimerExpired(iEnd) ) {
			return false;
		}
		xrtThreadYield();
	}
	return true;
}

/* Worker 任务：在客户端连接上执行三族同步发送。 */
typedef struct toursend {
	xwsconn* pConnection;
	xatomic32 Done;
	bool bOk;
} toursend;

static void tourSyncSendTask(xnetworker* pWorker, ptr pUserData)
{
	toursend* pTask = (toursend*)pUserData;
	xwsconn* pConnection = pTask->pConnection;
	xnetref Ref;
	bytes pOwned = NULL;
	xwswriter* pWriter = NULL;

	(void)pWorker;
	/* xnetresult 的 OK 是 0，不能用按位 &= 累积成功状态。 */
#define TOUR_SEND(Call) do { if ( (Call) != XNET_RESULT_OK ) goto Done; } while (0)
	TOUR_SEND(xrtWsConnText(pConnection, XRT_STR_LITERAL("t1")));
	TOUR_SEND(xrtWsConnBinary(pConnection, (xbytesview) { (cbytes)"b1", 2u }));
	TOUR_SEND(xrtWsConnSend(pConnection, XWS_OPCODE_TEXT,
		(xbytesview) { (cbytes)"s1", 2u }));
	/* Ref 仅在 OK 时接管；释放回调真正释放分配而非只计数。 */
	pOwned = (bytes)xrtMalloc(2u);
	if ( pOwned == NULL ) goto Done;
	memcpy(pOwned, "r1", 2u);
	Ref = (xnetref) { pOwned, 2u, tourRelease, NULL };
	TOUR_SEND(xrtWsConnTextRef(pConnection, &Ref));
	pOwned = (bytes)xrtMalloc(2u);
	if ( pOwned == NULL ) goto Done;
	memcpy(pOwned, "r2", 2u);
	Ref.Data = pOwned;
	TOUR_SEND(xrtWsConnBinaryRef(pConnection, &Ref));
	pOwned = (bytes)xrtMalloc(2u);
	if ( pOwned == NULL ) goto Done;
	memcpy(pOwned, "r3", 2u);
	Ref.Data = pOwned;
	TOUR_SEND(xrtWsConnSendRef(pConnection, XWS_OPCODE_TEXT, &Ref));
	pOwned = (bytes)xrtMalloc(2u);
	if ( pOwned == NULL ) goto Done;
	memcpy(pOwned, "k1", 2u);
	TOUR_SEND(xrtWsConnBinaryTake(pConnection, pOwned, 2u));
	pOwned = (bytes)xrtMalloc(2u);
	if ( pOwned == NULL ) goto Done;
	memcpy(pOwned, "k2", 2u);
	TOUR_SEND(xrtWsConnTextTake(pConnection, (str)pOwned, 2u));
	pOwned = (bytes)xrtMalloc(2u);
	if ( pOwned == NULL ) goto Done;
	memcpy(pOwned, "k3", 2u);
	TOUR_SEND(xrtWsConnSendTake(pConnection, XWS_OPCODE_BINARY, pOwned, 2u));
	pOwned = NULL;
	/* 两个 Take 分片组成第十条消息。 */
	pWriter = xrtWsConnBeginText(pConnection);
	if ( pWriter == NULL ) goto Done;
	pOwned = (bytes)xrtMalloc(4u);
	if ( pOwned == NULL ) goto Done;
	memcpy(pOwned, "frag", 4u);
	TOUR_SEND(xrtWsWriterWriteTake(pWriter, pOwned, 4u));
	pOwned = (bytes)xrtMalloc(3u);
	if ( pOwned == NULL ) goto Done;
	memcpy(pOwned, "end", 3u);
	TOUR_SEND(xrtWsWriterFinishTake(pWriter, pOwned, 3u));
	pOwned = NULL;
	if ( !xrtWsWriterIsFinished(pWriter) ) goto Done;
	TOUR_SEND(xrtWsConnPing(pConnection, (xbytesview) { (cbytes)"pk", 2u }));
	TOUR_SEND(xrtWsConnPong(pConnection, (xbytesview) { (cbytes)"pk", 2u }));
	pTask->bOk = true;
Done:
#undef TOUR_SEND
	xrtFree(pOwned);
	xrtWsWriterDestroy(pWriter);
	xrtAtomic32Store(&pTask->Done, 1, XMEMORY_RELEASE);
}

static void tourCloseTask(xnetworker* pWorker, ptr pData)
{
	toursend* pTask = (toursend*)pData;

	(void)pWorker;
	pTask->bOk = xrtWsConnClose(pTask->pConnection, 1000,
		XRT_STR_LITERAL("done")) == XNET_RESULT_OK;
	xrtAtomic32Store(&pTask->Done, 1, XMEMORY_RELEASE);
}

static bool tourFutureReady(xfuture* pFuture)
{
	return (pFuture != NULL) &&
		(xrtFutureWaitFor(pFuture, TOUR_DEADLINE_MS) == XWAIT_OK) &&
		(xrtFutureState(pFuture) == XFUTURE_RESOLVED);
}

/* 暂停断言必须失败可见，不能只打印“leaked”然后仍返回成功。 */
static bool tourHeld(tourstate* pState, uint32 iMessages)
{
	double Deadline = exampleTimerLimit(200);

	while ( !exampleTimerExpired(Deadline) ) {
		if ( (xrtAtomic32Load(&pState->Messages, XMEMORY_ACQUIRE) != iMessages) ||
			(xrtAtomic32Load(&pState->Errors, XMEMORY_ACQUIRE) != 0) ) return false;
		xrtThreadYield();
	}
	return true;
}

/* 握手失败/超时时，完成回调仍可能在收尾阶段才交付引用。 */
static void tourReleaseConnection(xatomicptr* pSlot)
{
	xwsconn* pConnection = (xwsconn*)xrtAtomicPtrExchange(pSlot, NULL, XMEMORY_ACQ_REL);

	if ( pConnection != NULL ) (void)xrtWsConnAbort(pConnection);
	xrtWsConnDestroy(pConnection);
}

int main(void)
{
	tourstate State;
	xnetengineconfig EngineConfig;
	xnetengine* pEngine = NULL;
	xhttpserverconfig ServerConfig;
	xhttpserverevents ServerEvents;
	xhttpserver* pServer = NULL;
	xhttpclientconfig ClientConfig;
	xhttpclient* pClient = NULL;
	xwsclientconfig WsConfig;
	xwsserverconfig WsServerConfig;
	xhttpcall* pCall = NULL;
	xhttprequest* pRequest = NULL;
	xhttprequest* pCloned = NULL;
	xwsconnconfig ConnConfig;
	xwsconn* pConn = NULL;
	xwsconn* pServerConn = NULL;
	xfuture* Futures[6] = { NULL };
	bytes pOwned = NULL;
	xnetref Ref;
	xnetaddr Address;
	toursend Send;
	char sUrl[128];
	char sKey[XWS_KEY_CAPACITY];
	int iLength;
	int iResult = 1;
	double Deadline;

	memset(&State, 0, sizeof(State));
	xrtAtomicPtrInit(&State.Client, NULL);
	xrtAtomicPtrInit(&State.Server, NULL);
	xrtAtomic32Init(&State.ClientDone, 0);
	xrtAtomic32Init(&State.ServerDone, 0);
	xrtAtomic32Init(&State.Messages, 0);
	xrtAtomic32Init(&State.Echoes, 0);
	xrtAtomic32Init(&State.Pings, 0);
	xrtAtomic32Init(&State.Pongs, 0);
	xrtAtomic32Init(&State.Closed, 0);
	xrtAtomic32Init(&State.CleanClosed, 0);
	xrtAtomic32Init(&State.Errors, 0);
	xrtAtomic32Init(&State.PauseNext, 0);
	xrtAtomic32Init(&g_Released, 0);
	memset(&Send, 0, sizeof(Send));
	xrtAtomic32Init(&Send.Done, 0);
	g_Events.MessageEnd = tourMessageEnd;
	g_Events.Ping = tourPing;
	g_Events.Pong = tourPong;
	g_Events.Close = tourClose;
	g_Events.Error = tourError;

	/* ---- 离线：请求构建器与配置校验。 ---- */
	pRequest = xrtWsRequestCreate(XRT_STR_LITERAL("wss://example.test/chat"));
	xrtWsClientConfigInit(&WsConfig);
	WsConfig.Protocols = XRT_STR_LITERAL("chat.v1, chat.v2");
	pCloned = xrtWsClientRequestCreate(
		XRT_STR_LITERAL("wss://example.test/chat"), &WsConfig, sKey);
	xrtWsConnConfigInit(&ConnConfig);
	xrtWsServerConfigInit(&WsServerConfig);
	WsServerConfig.Protocols = XRT_STR_LITERAL("chat.v1");
	if ( (pRequest == NULL) || (pCloned == NULL) ||
		!xrtWsKeyValid((xstrview) { sKey, XWS_KEY_SIZE }) ||
		xrtWsServerConfigValid(&(xwsserverconfig) { 0 }) ||
		!xrtWsServerConfigValid(&WsServerConfig) ||
		!xrtWsConnConfigValid(&ConnConfig) ||
		xrtWsConnConfigValid(&(xwsconnconfig) { 0 }) ||
		(xrtWsConnAttach(NULL, &ConnConfig, NULL, NULL) != NULL) ||
		(xrtWsConnAttachTls(NULL, &ConnConfig, NULL, NULL) != NULL) ) goto Cleanup;
	{
		xhttprequest* pClone = xrtWsClientRequestClone(pCloned, &WsConfig, State.Key);
		bool bValid = (pClone != NULL) &&
			xrtWsKeyValid((xstrview) { State.Key, XWS_KEY_SIZE });

		xrtHttpRequestDestroy(pClone);
		if ( !bValid ) goto Cleanup;
	}
	xrtHttpRequestDestroy(pCloned);
	pCloned = NULL;
	xrtHttpRequestDestroy(pRequest);
	pRequest = NULL;
	xrtClearError();  /* 以上负路径不能污染在线阶段的错误诊断。 */
	printf("conn-tour: offline request builders ok\n");

	/* ---- 在线：事件表和事件上下文必须同时绑定到客户端与服务端。 ---- */
	xrtNetEngineConfigInit(&EngineConfig);
	EngineConfig.Workers = TOUR_WORKERS;
	EngineConfig.Backend = TOUR_BACKEND;
	pEngine = xrtNetEngineCreate(&EngineConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) goto Cleanup;
	xrtHttpServerConfigInit(&ServerConfig);
	if ( !xrtNetAddrLoopback(&ServerConfig.Network.Listen.Address,
		XNET_FAMILY_IPV4, 0) ) goto Cleanup;
	xrtHttpServerEventsInit(&ServerEvents);
	ServerEvents.Request = tourRequest;
	ServerEvents.Data = &State;
	pServer = xrtHttpServerStart(pEngine, &ServerConfig, &ServerEvents);
	if ( (pServer == NULL) || !xrtHttpServerLocal(pServer, 0, &Address) ) goto Cleanup;
	xrtHttpClientConfigInit(&ClientConfig);
	pClient = xrtHttpClientCreate(pEngine, &ClientConfig);
	if ( pClient == NULL ) goto Cleanup;
	iLength = snprintf(sUrl, sizeof(sUrl), "ws://127.0.0.1:%u/chat", (unsigned)Address.Port);
	if ( (iLength < 0) || ((size_t)iLength >= sizeof(sUrl)) ) goto Cleanup;
	pCall = xrtWsConnect(pClient, (xstrview) { sUrl, (size_t)iLength },
		&WsConfig, &g_Events, &State, tourClientDone, &State);
	if ( (pCall == NULL) || !tourWait32Min(&State.ClientDone, 1) ||
		!tourWait32Min(&State.ServerDone, 1) ||
		(xrtAtomic32Load(&State.ClientDone, XMEMORY_ACQUIRE) != 1) ||
		(xrtAtomic32Load(&State.ServerDone, XMEMORY_ACQUIRE) != 1) ||
		!State.bIntroOk || !State.bClientCheckOk ) goto Cleanup;
	pConn = (xwsconn*)xrtAtomicPtrLoad(&State.Client, XMEMORY_ACQUIRE);
	pServerConn = (xwsconn*)xrtAtomicPtrLoad(&State.Server, XMEMORY_ACQUIRE);
	if ( (pConn == NULL) || (pServerConn == NULL) ) goto Cleanup;
	printf("conn-tour: live pair upgraded, introspection ok\n");

	/* ---- 同步发送九形态 + Writer-Take：上下文留在 main 作用域。 ---- */
	Send.pConnection = pConn;
	if ( !xrtNetEnginePost(pEngine, xrtNetWorkerIndex(xrtWsConnWorker(pConn)),
			tourSyncSendTask, &Send) ||
		!tourWait32Min(&Send.Done, 1) || !Send.bOk ||
		!tourWait32Min(&State.Messages, 10) || !tourWait32Min(&State.Echoes, 10) ) goto Cleanup;
	printf("conn-tour: sync send x9 + writer-take delivered ok\n");

	/* ---- 异步发送：从主线程逐一检查 Future 的成功终态。 ---- */
	Futures[0] = xrtWsConnTextAsync(pConn, XRT_STR_LITERAL("a1"));
	Futures[1] = xrtWsConnBinaryAsync(pConn, (xbytesview) { (cbytes)"a2", 2u });
	pOwned = (bytes)xrtMalloc(2u);
	if ( pOwned == NULL ) goto Cleanup;
	memcpy(pOwned, "a3", 2u);
	Ref = (xnetref) { pOwned, 2u, tourRelease, NULL };
	Futures[2] = xrtWsConnTextRefAsync(pConn, &Ref);
	if ( Futures[2] == NULL ) goto Cleanup;
	pOwned = NULL;  /* 非空 Future 已受理，立即转移 Ref 所有权。 */
	Futures[3] = xrtWsConnSendAsync(pConn, XWS_OPCODE_TEXT,
		(xbytesview) { (cbytes)"a4", 2u });
	Futures[4] = xrtWsConnPingAsync(pConn, (xbytesview) { (cbytes)"pk", 2u });
	Futures[5] = xrtWsConnWaitAsync(pConn, XWS_CONN_WAIT_WRITE);
	for ( size_t i = 0; i < 6; i++ ) {
		if ( !tourFutureReady(Futures[i]) ) goto Cleanup;
		xrtFutureDestroy(Futures[i]);
		Futures[i] = NULL;
	}
	if ( !tourWait32Min(&State.Messages, 14) || !tourWait32Min(&State.Echoes, 14) ||
		!tourWait32Min(&State.Pongs, 2) || !tourWait32Min(&State.Pings, 2) ||
		!tourWait32Min(&g_Released, 4) ||
		(xrtAtomic32Load(&g_Released, XMEMORY_ACQUIRE) != 4) ||
		(xrtWsConnState(pConn) != XWS_CONN_OPEN) ||
		(xrtWsConnState(pServerConn) != XWS_CONN_OPEN) ) goto Cleanup;
	printf("conn-tour: async family + wait barrier ok\n");
	printf("conn-tour: pong observed via auto-pong ok\n");

	/* ---- 主线程 Pause 与回调内 Pause 分别验证，控制帧不代替 Resume。 ---- */
	xrtWsConnPause(pServerConn);
	Futures[0] = xrtWsConnTextAsync(pConn, XRT_STR_LITERAL("paused-main"));
	if ( !xrtWsConnPaused(pServerConn) || !tourFutureReady(Futures[0]) ||
		!tourHeld(&State, 14) || !xrtWsConnResume(pServerConn) ||
		!tourWait32Min(&State.Messages, 15) || !tourWait32Min(&State.Echoes, 15) ) goto Cleanup;
	xrtFutureDestroy(Futures[0]);
	Futures[0] = NULL;
	xrtAtomic32Store(&State.PauseNext, 1, XMEMORY_RELEASE);
	Futures[0] = xrtWsConnTextAsync(pConn, XRT_STR_LITERAL("pause-in-callback"));
	Futures[1] = xrtWsConnTextAsync(pConn, XRT_STR_LITERAL("after-callback"));
	if ( !tourFutureReady(Futures[0]) || !tourFutureReady(Futures[1]) ||
		!tourWait32Min(&State.Messages, 16) || !xrtWsConnPaused(pServerConn) ||
		!tourHeld(&State, 16) || !xrtWsConnResume(pServerConn) ||
		!tourWait32Min(&State.Messages, 17) || !tourWait32Min(&State.Echoes, 17) ) goto Cleanup;
	for ( size_t i = 0; i < 2; i++ ) {
		xrtFutureDestroy(Futures[i]);
		Futures[i] = NULL;
	}
	Futures[0] = xrtWsConnWaitAsync(pConn, XWS_CONN_WAIT_WRITE);
	if ( !tourFutureReady(Futures[0]) || (xrtWsConnPending(pConn) != 0) ||
		(xrtWsConnAsyncBytes(pConn) != 0) || (xrtWsConnAsyncCount(pConn) != 0) ) goto Cleanup;
	xrtFutureDestroy(Futures[0]);
	Futures[0] = NULL;
	printf("conn-tour: pause main/callback held, resume delivered ok\n");

	/* ---- 双方干净关闭，观察 Close 回调和 CloseInfo 而非假定成功。 ---- */
	Send.bOk = false;
	xrtAtomic32Store(&Send.Done, 0, XMEMORY_RELEASE);
	if ( !xrtNetEnginePost(pEngine, xrtNetWorkerIndex(xrtWsConnWorker(pConn)),
			tourCloseTask, &Send) || !tourWait32Min(&Send.Done, 1) || !Send.bOk ||
		!tourWait32Min(&State.Closed, 2) ||
		(xrtAtomic32Load(&State.CleanClosed, XMEMORY_ACQUIRE) != 2) ||
		(xrtAtomic32Load(&State.Errors, XMEMORY_ACQUIRE) != 0) ||
		(xrtAtomic32Load(&State.Messages, XMEMORY_ACQUIRE) != 17) ||
		(xrtAtomic32Load(&State.Echoes, XMEMORY_ACQUIRE) != 17) ||
		(xrtWsConnError(pConn) != NULL) || (xrtWsConnError(pServerConn) != NULL) ) goto Cleanup;
	printf("conn-tour: close handshake clean on both peers ok\n");
	iResult = 0;

Cleanup:
	xrtFree(pOwned);
	for ( size_t i = 0; i < 6; i++ ) xrtFutureDestroy(Futures[i]);
	xrtHttpRequestDestroy(pCloned);
	xrtHttpRequestDestroy(pRequest);
	if ( pCall != NULL ) (void)xrtHttpCallCancel(pCall);
	/* 回调可能在失败收尾中才完成；先释放已经交付的连接引用。 */
	tourReleaseConnection(&State.Client);
	tourReleaseConnection(&State.Server);
	xrtHttpCallDestroy(pCall);
	xrtHttpClientDestroy(pClient);
	/* Server 的 Destroy 只释放引用，不隐式停止监听。 */
	if ( pServer != NULL ) (void)xrtHttpServerAbort(pServer);
	xrtHttpServerDestroy(pServer);
	/* main 内的 Send / State 在停止 Worker 的整个过程中保持有效。 */
	Deadline = exampleTimerLimit(TOUR_DEADLINE_MS);
	/* 上层 Destroy 会投递异步关闭；最后一个内部对象释放后才能 Stop。 */
	while ( (pEngine != NULL) && !xrtNetEngineStop(pEngine) ) {
		tourReleaseConnection(&State.Client);
		tourReleaseConnection(&State.Server);
		if ( exampleTimerExpired(Deadline) ) { iResult = 1; break; }
		xrtThreadYield();
	}
	tourReleaseConnection(&State.Client);
	tourReleaseConnection(&State.Server);
	if ( (pEngine != NULL) && !xrtNetEngineDestroy(pEngine) ) iResult = 1;
	if ( iResult != 0 ) fprintf(stderr, "conn-tour: verification failed\n");
	return iResult;
}

$ gcc -O1 -DXRT_MODULE_ALL -DXHTTP_MODULE_ALL -DXWS_MODULE_ALL -I single -I single/extlibs -include xrt.h -include xhttp.h -include xws.h impl.c extlibs/xws/examples/websocket/connection_tour/main.c -lws2_32 -lbcrypt -ladvapi32 -liphlpapiconn-tour: offline request builders okconn-tour: live pair upgraded, introspection okconn-tour: sync send x9 + writer-take delivered okconn-tour: async family + wait barrier okconn-tour: pong observed via auto-pong okconn-tour: pause main/callback held, resume delivered okconn-tour: close handshake clean on both peers ok

刚才发生了什么。 七行即七类契约。① 离线构建：不建连接也能构造请求/应答材料（握手层的纯函数面——第 94 章）。② 实连接对：xhttp 服务器与客户端经环回完成升级，自省（角色/协议/状态）在连接所属 Worker 回调内执行——同步调用的 Worker 归属纪律。③ 九种同步发送 + writer-take：copy/ref/take × text/binary 加 writer 路径（第 111 章）全部投递成功——Take 负载的释放回调恰好一次（所有权终态验证）。④ 异步族 + 等待栅栏：Future 发送与等待（取消不销毁连接）。⑤ 自动 Pong：Ping 观察到自动应答——控制帧自动化。⑥ 暂停/恢复：主回调暂停接收、数据保持、恢复后投递——流控语义。⑦ 干净关闭：双方 Close 握手完成。一行一约、七行全绿——本章的契约表就是这张输出。

第二个完整程序：引用发送的释放契约

第二个程序来自 examples/websocket/connection_ref——ref 路径的所有权模板：

extlibs/xws/examples/websocket/connection_ref/main.c源码 ↗
#include <stdio.h>
#include <string.h>

#include <xws.h>



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



/* 在 Connection 所属 Worker 上发送一个可延迟释放的业务负载。 */
static xnetresult sendPayload(
	xwsconn* 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 = xrtWsConnBinaryRef(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;
}

刚才发生了什么。 ① xnetref 四元组{指针, 长度, releasePayload 回调, 上下文}——负载由 xrtMalloc 独立分配（不借用调用方缓冲）。② xrtWsConnBinaryRef(连接, &Ref) 引用入队——排队期间零复制、写完 socket 后 release 恰好一次；发送失败（未受理）时所有权归还——示例的失败分支自己 xrtFree（契约：未受理归调用方）。③ 本示例是“由已连接 Worker 回调调用”的函数模板（main 只验证就绪）——真实使用点在消息/路由回调里。第 96 章讲过核心库的 ref 发送；这里是连接层的同一语义——词汇完全通用。

110.4契约

接管边界：xwsconn 接管已握手 TCP/TLS 流；不管 URL/DNS/HTTP/证书/重定向——上下两层各归其位。
事件模型：消息开始/数据/结束 + Ping/Pong/Close/错误/终态；数据视图仅回调内有效；角色建立时固定。
流控：pause/resume；不允许无界应用队列；发送继承水位与 AGAIN 契约。
所有权原子：copy/ref/take 三态；未受理归调用方、受理后连接释放恰好一次。
Future 桥：只包已建立连接的操作；取消等待≠销毁连接；协程走通用 Future/Wait——无第二调度器。
关闭顺序：Close → 停新数据 → 等对端/截止 → 关写方向 → 等传输；abort 也保证引用释放。
错误域：协议错误映射 WS 域码+RFC Close code；网络/TLS/应用错误保留各自域——不压扁。
裁剪：连接/所有权/异步/TLS/分组分表独立（帧层不带入网络——核心裁剪表的延续）。

常见错误

坑 1：pause 之后忘了 resume（或对端超时）

症状：客户端“莫名其妙收不到消息”——服务端 pause 后消费逻辑挂了，连接半死不活到超时。

原因：pause 是“暂时不收”——必须与一个明确的恢复路径配对（异步消费者完成、下游就绪）。挂起期间对端视角就是连接卡死。

错误写法

onMessage(...) {
	if ( busy() ) {
		xrtWsConnPause(pConn);   /* 暂停后无人恢复 */
	}
}

正确写法

onMessage(...) {
	if ( busy() ) {
		xrtWsConnPause(pConn);
		defer_resume_on_ready(pConn);   /* 明确的恢复触发点 */
	}
}

坑 2：把 Future 取消当连接关闭

症状：等发送 Future 超时后直接 Destroy 连接——对端正在收的半条消息被腰斩，还可能双释放。

原因：取消等待只是“不等了”——连接和已受理数据的所有权完好。按操作契约决定：继续用（等下一个 Future）、正常 Close、或 abort。

错误写法

if ( wait_timeout(pSendFuture) ) {
	xrtWsConnDestroy(pConn);   /* 取消≠销毁：可能双释放/半消息 */
}

正确写法

if ( wait_timeout(pSendFuture) ) {
	/* 数据是否已受理？连接状态查得到——再决定关闭方式 */
	if ( !xrtWsConnClose(pConn, Code, Reason) ) {
		xrtWsConnAbort(pConn);   /* 关闭不行才中止 */
	}
}

坑 3：在非 Worker 线程调同步发送

症状：偶发状态错乱——同步发送设计在连接所属 Worker 上执行（巡检第 2 行的注释明说）。

原因：连接对象非多线程共享的可变状态——跨线程操作走 Future/异步族（它们内部正确投递）。这与第 86 章会话层、第 104 章服务端事件的 Worker 串行纪律同源。

错误写法

/* 任意线程直接调同步发送 */
xrtWsConnText(pConn, Msg);   /* 非所属 Worker：竞态 */

正确写法

/* 跨线程：走 Future/异步族（内部投递到连接 Worker） */
/* 形态见 websocket_http_future 模块：ConnectAsync 一族的等待语义 */

编程练习

基础：巡检逐行标注

跑通 connection_tour，给七行输出各写一句“验证了哪条契约”。验收标准：七条契约全部能指到本章契约区的对应条目。

进阶：暂停-恢复压力对

基于巡检的回环骨架：消费端随机 pause 0-50ms 后 resume，发送端全速发 1000 条——统计投递与丢失。验收标准：零丢失（暂停只是延迟）；内存恒定（第 6 章统计）；对端视角只是吞吐波动。

挑战：双通道代理

连接 A 收到的消息转发到连接 B（反向亦然）——双连接的 pause 联动：B 忙时 pause A。验收标准：转发零丢失；B 慢时 A 的接收同步暂停（内存恒定）；任一端关闭时另一端干净收尾（Close 传播）。

小结

知识点	速查
分工边界	核心库（95-97 章协议零件）+ xws（连接/组/路由工程层）
接管形态	已握手 TCP/TLS 流；不管 URL/DNS/HTTP/证书
事件模型	消息三段 + Ping/Pong/Close/错误/终态；视图仅回调内有效
流控	pause/resume；无界队列禁止；AGAIN 结果+水位同全库
所有权三态	copy/ref/take；未受理归调用方；受理后释放恰好一次
Future 桥	只包已建立连接操作；取消≠销毁；协程走通用体系
关闭顺序	Close→停发→等对端→关写→等传输；abort 保引用释放
错误域	WS 域码+RFC Close code；网络/TLS/应用各留各域
巡检	connection_tour 七行=七类契约的可执行表
裁剪	连接/所有权/异步/TLS/分组独立分表

上一章第 109 章服务端收官：压缩、Range 与综合下一章第 111 章 xws（中）：发送路径——writer、引用与压缩

📖 本章涉及的 API 参考手册：xws-websocket_runtime（67 个条目）xws-websocket_http（14 个条目）net（247 个条目）

点击查看完整签名、参数约束、返回值与错误说明