Chapter 110 xws (Part 1): Connection Management and the Full-Chain Tour

Chapter Orientation

Volume 9's Chapters 95–97 covered WebSocket's protocol core (frames/messages/streams); the xws extension library completes the engineering connection layer on top: xwsconn takes over a TCP or TLS stream whose handshake is already done — it is not responsible for URL, DNS, HTTP requests, certificate policy, or redirects (those belong to Chapter 94's upgrade and Volume 8's TLS); it manages only the connection's message events, backpressure, close, and introspection. Three blocks in this chapter: events and view discipline (message start/data/end, Ping/Pong/Close events; data views valid only inside callbacks); flow control and the send contract (pause/resume receive flow control, high/low watermarks and the AGAIN result — inheriting the network stream's full vocabulary); the full-chain tour (connection_tour's seven self-check lines covering offline building, a live connection pair, nine send kinds, the async family, auto-Pong, pause/resume, and a clean close — this chapter's executable contract table). The Future/coroutine bridge and the close protocol close the chapter.

110.1引入

xws's division of labor is worth drawing first: the core library (Volume 9) provides frames, messages, the handshake, extension negotiation, permessage-deflate — composable protocol parts; the xws extension layer (from this chapter) wires those parts onto real connections and adds engineering facilities (connection events, reference lifecycle, Future waiting, connection groups, server routing). The contract docs state plainly that "HTTP client/server adapters, routing, and framework-style convenience objects have left the core" — the extension layer is positioned as the rebuilding zone for high-level capabilities, and must not let the object model pollute the frame and direct send/receive paths backward. This "thin core, thick extension" layering is fully isomorphic to xhttp (protocol core in Volume 9, application layer in Volume 10).

The connection layer's first design decision is take-over, not creation: xwsconn receives "a stream whose handshake is already complete" — who completed the handshake (Chapter 94's upgrade flow, xws server routing (Chapter 113)) does not matter; the connection layer faces only "a ready transport". This lets it stand equally above TCP and TLS (Chapter 97's TLS take-over path).

110.2概念

Event model and view discipline

Connection events: message start, data fragment, message end, Ping, Pong, Close, error, final close — more than Chapter 96's Stream layer, Ping/Pong get independent events (control frames are not only auto-answered; they are also visible). View discipline matches the whole library: data views in a callback are valid only during the callback; saving across callbacks requires copying or taking over an owning buffer. The role rules inherit from the core layer: the server requires client frames masked, the client requires server frames unmasked — the role is fixed when the connection is established, never guessed per frame.

Flow control: pause/resume and watermarks

1Receivethe message event stream -> application consumption

2Slow consumptionpause suspends receiving -> the peer feels TCP backpressure -> memory stays flat

3Resumeresume continues - never solve a slow consumer with an infinitely growing application queue

4Sendinherits the network stream's high/low watermarks, max queued bytes, and the AGAIN result contract

This is the connection-layer implementation of the "stop consuming = stop receiving" discipline (Chapter 62's skeleton economics). The send path's AGAIN and watermark vocabulary is identical to Chapter 96 — having learned the Stream layer, there are zero new words here.

The three send-ownership states (a preview of the send path)

copy (copied before return — short control frames/short messages), ref (release called exactly once when the queue drains it — static or shared large blocks), take/buffer (handing over already-allocated memory or a network buffer chain) — Chapter 111 expands the writer forms; this chapter first establishes the atomic rule "before acceptance ownership stays with the caller; after acceptance only the connection releases".

Future and coroutine bridge

Futures wrap only sends, closes, drains, and final-state waits on an already-established connection — no hidden HTTP client is created. Two disciplines: cancellation of a wait ≠ destroying the connection — the caller decides per the specific operation's contract whether to keep using it, close normally, or abort; coroutines wait through the generic Future/Wait system (Chapter 58) — no second scheduler is maintained inside WebSocket. The connection_future example family demonstrates the waiting forms.

Close protocol and error domains

Normal close order: send or answer Close → stop sending new data → wait for the peer's Close or the deadline → shut the write direction → wait for the transport to end. Protocol violation, limit excess, and underlying failure may abort directly — but every accepted reference must be released exactly once (the terminal-state guarantee of reference ownership). Error model: protocol errors use stable WebSocket-domain error codes and map to RFC Close codes where possible; network/TLS/application-callback errors keep their own error domains — never flattened into one boolean cause (Chapter 4's cause chain honored at the protocol layer).

110.3示例

First complete program: the seven-line full-chain tour

The program below is from examples/websocket/connection_tour — a full-feature self-check of the connection layer after an xhttp server and client loop back into each other:

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

What just happened. Seven lines, seven contract classes. (1) Offline building: request/response materials can be constructed without a connection (the handshake layer's pure-function face — Chapter 94). (2) Live connection pair: an xhttp server and client complete the upgrade over loopback; introspection (role/protocol/state) runs inside the owning Worker's callback — the Worker-affinity discipline of synchronous calls. (3) Nine synchronous sends + writer-take: copy/ref/take × text/binary plus the writer path (Chapter 111) all deliver — the Take payload's release callback fires exactly once (terminal-state ownership verification). (4) Async family + wait barrier: Future sends and waits (cancelling does not destroy the connection). (5) Auto-Pong: a Ping observes the automatic answer — control-frame automation. (6) Pause/resume: the main callback pauses receiving, data is held, delivery resumes after — flow-control semantics. (7) Clean close: both peers complete the Close handshake. One line, one contract; seven lines, all green — this chapter's contract table is exactly that output.

Second complete program: the release contract of reference sending

The second program is from examples/websocket/connection_ref — the ownership template of the ref path:

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;
}

What just happened. (1) The xnetref quadruple {pointer, length, releasePayload callback, context} — the payload is independently allocated by xrtMalloc (not borrowing the caller's buffer). (2) xrtWsConnBinaryRef(连接, &Ref) (connection) queues the reference — zero copies while queued, release exactly once after the socket write; on send failure (not accepted) ownership returns — the sample's failure branch calls xrtFree itself (contract: unaccepted stays with the caller). (3) This sample is a function template "called from a connected Worker's callback" (main only verifies readiness) — the real call sites live in message/routing callbacks. Chapter 96 covered the core library's ref sending; this is the same semantics at the connection layer — the vocabulary is fully shared.

110.4契约

Take-over boundary: xwsconn takes over handshaked TCP/TLS streams; it does not manage URL/DNS/HTTP/certificates/redirects — the layers above and below each stay in place.
Event model: message start/data/end + Ping/Pong/Close/error/final state; data views valid only inside callbacks; role fixed at establishment.
Flow control: pause/resume; unbounded application queues forbidden; sending inherits the watermark and AGAIN contracts.
Ownership atomicity: copy/ref/take three states; unaccepted stays with the caller, after acceptance the connection releases exactly once.
Future bridge: wraps only operations on established connections; cancelling a wait ≠ destroying the connection; coroutines go through the generic Future/Wait — no second scheduler.
Close order: Close → stop new data → wait for peer/deadline → shut write direction → wait for transport; abort also guarantees reference release.
Error domains: protocol errors map to WS-domain codes + RFC Close codes; network/TLS/application errors keep their own domains — never flattened.
Trimming: connection/ownership/async/TLS/grouping in independent tables (the frame layer pulls in no networking — a continuation of the core trimming table).

Common Mistakes

Pitfall 1: pause without resume (or the peer times out)：Pitfall 1: pause without resume (or the peer times out)

Symptom: the client "mysteriously stops receiving messages" — the server paused, the consuming logic stalled, and the connection half-lives until timeout.

Cause: pause means "not receiving for now" — it must be paired with an explicit resume path (async consumer finished, downstream ready). While suspended, the peer's view is a stuck connection.

错误写法

onMessage(...) {
	if ( busy() ) {
		xrtWsConnPause(pConn);   /* paused, nobody resumes */
	}
}

正确写法

onMessage(...) {
	if ( busy() ) {
		xrtWsConnPause(pConn);
		defer_resume_on_ready(pConn);   /* an explicit resume trigger point */
	}
}

Pitfall 2: treating a Future cancel as a connection close：Pitfall 2: treating a Future cancel as a connection close

Symptom: after a send Future wait times out, the connection is Destroyed outright — the half message the peer is receiving gets cut in two, and a double release is possible.

Cause: cancelling a wait only means "stop waiting" — the connection and the ownership of accepted data are intact. Decide per the operation's contract: keep using it (wait on the next Future), close normally, or abort.

错误写法

if ( wait_timeout(pSendFuture) ) {
	xrtWsConnDestroy(pConn);   /* cancel != destroy: possible double release / half message */
}

正确写法

if ( wait_timeout(pSendFuture) ) {
	/* was the data accepted? the connection state tells you - then choose the close style */
	if ( !xrtWsConnClose(pConn, Code, Reason) ) {
		xrtWsConnAbort(pConn);   /* abort only when close fails */
	}
}

Pitfall 3: calling synchronous sends from a non-Worker thread：Pitfall 3: calling synchronous sends from a non-Worker thread

Symptom: occasional state scrambling — synchronous sends are designed to run on the connection's owning Worker (the tour's line-2 comment says so explicitly).

Cause: the connection object is not multithread-shared mutable state — cross-thread operations go through the Future/asynchronous family (they dispatch correctly inside). The same source as Chapter 86's session layer and Chapter 104's server-event Worker-serial discipline.

错误写法

/* a synchronous send called directly from an arbitrary thread */
xrtWsConnText(pConn, Msg);   /* not the owning Worker: a race */

正确写法

/* cross-thread: go through the Future/async family (dispatched to the connection's Worker internally) */
/* see the websocket_http_future module: the waiting semantics of the ConnectAsync family */

Programming Exercises

Basic: annotate the tour line by line

Run connection_tour and write one sentence per output line stating "which contract it verified". Acceptance criteria: all seven contracts can be pointed at the corresponding entry in this chapter's Contracts section.

Advanced: a pause-resume stress pair

On the tour's loopback skeleton: the consumer randomly pauses 0–50 ms then resumes, the sender drives a 1000-message workload at full speed — tally deliveries and losses. Acceptance criteria: zero loss (pause only delays); memory constant (Chapter 6 stats); from the peer's view, only throughput fluctuates.

Challenge: a dual-channel proxy

Messages received on connection A are forwarded to connection B (and vice versa) — with pause linkage across the two connections: when B is busy, pause A. Acceptance criteria: zero forwarding loss; when B slows, A's receiving pauses in step (memory constant); when either side closes, the other finishes cleanly (Close propagates).

Summary

Topic	Quick reference
Division of labor	core library (protocol parts, Chapters 95–97) + xws (the connection/group/routing engineering layer)
Take-over form	a handshaked TCP/TLS stream; no URL/DNS/HTTP/certificates
Event model	three message phases + Ping/Pong/Close/error/final; views valid only inside callbacks
Flow control	pause/resume; unbounded queues forbidden; AGAIN result + watermarks as everywhere
Three ownership states	copy/ref/take; unaccepted stays with the caller; after acceptance released exactly once
Future bridge	wraps only established-connection operations; cancel ≠ destroy; coroutines go through the generic system
Close order	Close → stop sending → wait peer → shut write → wait transport; abort still guarantees reference release
Error domains	WS-domain codes + RFC Close codes; network/TLS/application each keep their domain
The tour	connection_tour's seven lines = an executable table of seven contract classes
Trimming	connection/ownership/async/TLS/grouping in independent tables

PreviousCh. 109 服务端收官：压缩、Range 与综合 NextCh. 111 xws（中）：发送路径——writer、引用与压缩

📖 API references in this chapter:xws-websocket_runtime (67 entries)xws-websocket_http (14 entries)net (247 entries)

Click for full signatures, parameter constraints, return values and error notes