第 87 章 TLS 流：TCP 上的组合层

本章导读

第 86 章的会话机自己不碰网络；第 82 章的客户端拨号已经偷偷用了 Stream 层。本章把 Stream 层（xtlsstream）正面讲完：服务端形态——TCP Accept 回调里 xrtTlsStreamAccept 一步建立"TCP 流 + TLS 会话机"的组合对象，之后业务只看到明文读写；事件模型——Open/Read/Writable/Close 四回调（第 82 章客户端同一套）；背压——明文接收 Available + 短写等待 Writable，慢客户端不会撑爆服务端内存；生命周期——共享配置引用计数，监听器与全部连接关闭后统一释放。这是全库"骨架组合"模式的又一次兑现：Engine（第 64 章）+ Listener/Stream（第 67 章）+ TLS 会话机（第 86 章）拼成一个可 openssl s_client 直连的完整服务。

87.1引入

把第 86 章的回显对打变成真实服务，中间差的只是"传输接线"：Accept 到 TCP 流 → 建组合对象 → 握手自动跑 → 事件回调送明文。听起来薄薄一层，但每一处都是坑位：握手期间的字节谁喂？写背压卡住时明文消费停在哪？连接关闭与监听器关闭怎么协调才不泄漏？Stream 层把这层接线标准化——你写的是 Echo 回调里的业务逻辑，生命周期与背压由组合对象管理。

服务的形态也值得看：一个进程、一个 Engine、一个 Listener、N 条 TLS 连接共享同一份服务端配置（身份/ALPN）与事件表——配置引用计数让"装配一次、N 连接复用"零拷贝。这与第 83 章身份的共享、第 85 章验证器的共享连成一条线：TLS 的全部重对象都是不可变共享值。

87.2概念

组合对象的建立：Accept 与 Dial 两侧

服务端：xrtTlsStreamAccept(TCP流, 服务端配置, 流配置, 事件表, 用户数据, &流)——在 TCP Accept 回调内一步建立组合对象；握手自动跑，成功后 Open 回调触发。
客户端：xrtTlsDial/DialAsync（第 82 章）内部就是"TCP 拨号 + StreamAttach 客户端方向"——客户端没有独立的 Attach 范例因为拨号已内联。

两侧共享 xtlsstream 类型与全部应用接口——Send/Available/Buffer/Consume/Close/Abort/State（第 82 章速查）。差异只在建立方式与方向。

事件模型与硬背压

四回调与 TCP 流（第 67 章）完全同构：Open（握手完成/READY）、Read（明文到达）、Writable（发送预算恢复）、Close（唯一终态：xnetresult + 结构化错误）。硬背压的机制：明文接收区有水位（Available 非零即待处理），业务没消费完之前接收侧不再推进——回显服务把"消费明文"与"发回明文"绑在同一循环：逐 Span Send，短写（XTLS_AGAIN）就暂停消费等 Writable 再续——慢客户端的 TCP 窗口收缩 → TLS 记录发不出去 → 明文消费暂停 → 接收水位不动 → 内存恒定。这就是第 62 章"泵零阻塞"承诺在 TLS 上的兑现。

配置共享与生命周期

1装配身份（第 83 章）+ 服务端配置 + 流配置 + 事件表 —— 全部栈上或共享对象

2接入每条 TCP 连接 StreamAccept 一次，组合对象持有配置引用

3运行N 连接并发，回调在 Engine Worker 上执行（第 64 章）

4关闭每连接 Close 回调递减计数；监听器关闭置位；两者归零后共享对象统一释放

共享配置在 Listener 与全部连接关闭前不可释放——示例用原子计数（Connections/ListenerClosed）编排这个汇合点。单连接的规则与第 82 章一致：Close/Abort → 等 XTLS_STREAM_CLOSED/FAILED → Destroy。

TLS 特有的接线细节

三个值得知道的实现衔接：握手字节由组合层自动喂（你从没调过 Feed——StreamAccept 后握手就是"传输的事"）；传输查询xrtTlsStreamTransport(流) 借出底层 TCP 流——Open 回调里查对端地址（xrtNetStreamRemote）打日志是标准动作；两级关闭——StreamClose 走 close_notify 排空（第 82 章），Close 回调在组合对象终态时触发一次。配置分两份：xtlsserverconfig（协议面：身份/ALPN）与 xtlsstreamconfig（流面：缓冲水位等）——协议与传输的关注点分离到配置层。

87.3示例

第一个完整程序：事件驱动 TLS 回显服务

下面的程序来自 examples/tls/stream/main.c——可用 openssl s_client 直连的完整 TLS 1.3 Echo 服务：

examples/tls/stream/main.c源码 ↗
#include "../common.h"























 */


/* Echo 服务共享配置只在 Listener 和全部连接关闭后释放。 */
typedef struct example_tls_stream {
	xtlsserverconfig ServerConfig;
	xtlsstreamconfig StreamConfig;
	xtlsstreamevents StreamEvents;
	xatomic32 Connections;
	xatomic32 ListenerClosed;
} example_tls_stream;



/* 输出当前线程保存的结构化根错误。 */
static void exampleTlsStreamError(cstr sPrefix)
{
	const xerror* pError = xrtGetError();

	fprintf(
		stderr,
		"%s: %s\n",
		sPrefix,
		pError != NULL ? xrtErrorMessage(pError) : "unknown error"
	);
}



/* 从借用明文块逐 Span 回送；短写等待 Writable 后继续。 */
static void exampleTlsStreamEcho(xtlsstream* pStream)
{
	while ( xrtTlsStreamAvailable(pStream) != 0 ) {
		const xnetbuf* pBuffer = xrtTlsStreamBuffer(pStream);
		xnetspan Span;
		size_t iWritten = 0;
		xtlsresult Result;

		if ( (pBuffer == NULL) || !xrtNetBufFront(pBuffer, &Span) ) {
			(void)xrtTlsStreamAbort(pStream);
			return;
		}
		Result = xrtTlsStreamSend(
			pStream,
			Span.Data,
			Span.Size,
			&iWritten
		);
		if ( (iWritten != 0) &&
			!xrtTlsStreamConsume(pStream, iWritten) ) {
			(void)xrtTlsStreamAbort(pStream);
			return;
		}
		if ( Result == XTLS_AGAIN ) {
			return;
		}
		if ( Result != XTLS_OK ) {
			(void)xrtTlsStreamAbort(pStream);
			return;
		}
	}
}



/* TLS READY 后记录活动连接。 */
static void exampleTlsStreamOpen(xtlsstream* pStream, ptr pData)
{
	example_tls_stream* pExample = (example_tls_stream*)pData;
	xnetaddr Remote;
	char Address[128];

	(void)xrtAtomic32FetchAdd(
		&pExample->Connections,
		1,
		XMEMORY_RELEASE
	);
	if ( xrtNetStreamRemote(
		xrtTlsStreamTransport(pStream),
		&Remote
	) && xrtNetAddrText(
		&Remote,
		Address,
		sizeof(Address)
	) ) {
		printf("TLS client open: %s\n", Address);
	}
}



/* 新明文和写背压解除都进入同一个无重复 Echo 路径。 */
static void exampleTlsStreamRead(
	xtlsstream* pStream,
	const xnetbuf* pBuffer,
	ptr pData
)
{
	(void)pBuffer;
	(void)pData;
	exampleTlsStreamEcho(pStream);
}



/* 写入队列重新可用时继续尚未消费的借用明文。 */
static void exampleTlsStreamWritable(xtlsstream* pStream, ptr pData)
{
	(void)pData;
	exampleTlsStreamEcho(pStream);
}



/* 记录组合连接终态；失败根因只在回调期间打印。 */
static void exampleTlsStreamClose(
	xtlsstream* pStream,
	xnetresult Result,
	const xerror* pError,
	ptr pData
)
{
	example_tls_stream* pExample = (example_tls_stream*)pData;

	(void)pStream;
	if ( (Result != XNET_RESULT_OK) && (pError != NULL) ) {
		fprintf(
			stderr,
			"TLS client closed: %s\n",
			xrtErrorMessage(pError)
		);
	}
	(void)xrtAtomic32FetchSub(
		&pExample->Connections,
		1,
		XMEMORY_RELEASE
	);
}



/* 在 TCP Accept 回调内一步建立服务端 TLS 组合对象。 */
static bool exampleTlsStreamAccept(
	xnetlistener* pListener,
	xnetstream* pTransport,
	ptr pData
)
{
	example_tls_stream* pExample = (example_tls_stream*)pData;
	xtlsstream* pStream = NULL;
	bool bAccepted;

	(void)pListener;
	bAccepted = xrtTlsStreamAccept(
		pTransport,
		&pExample->ServerConfig,
		&pExample->StreamConfig,
		&pExample->StreamEvents,
		pExample,
		&pStream
	);
	if ( bAccepted ) {
		xrtTlsStreamDestroy(pStream);
	}
	return bAccepted;
}



/* 记录 Listener 唯一关闭回调。 */
static void exampleTlsStreamListenerClose(
	xnetlistener* pListener,
	ptr pData
)
{
	example_tls_stream* pExample = (example_tls_stream*)pData;

	(void)pListener;
	xrtAtomic32Store(
		&pExample->ListenerClosed,
		1,
		XMEMORY_RELEASE
	);
}



/* 运行一个可由 openssl s_client 访问的 TLS 1.3 Echo 服务。 */
int main(int argc, char** argv)
{
	static const xstrview Protocols[] = {
		XRT_STR_INIT("http/1.1")
	};
	example_tls_stream Example;
	uint8* pCertificateData = NULL;
	uint8* pPrivateData = NULL;
	size_t iCertificateSize = 0;
	size_t iPrivateSize = 0;
	xbytesview Certificate;
	xtlsidentity* pIdentity = NULL;
	xnetengineconfig EngineConfig;
	xnetlistenconfig ListenConfig;
	xnetlistenerevents ListenerEvents;
	xnetengine* pEngine = NULL;
	xnetlistener* pListener = NULL;
	xnetaddr Address;
	unsigned long iPort = 8443u;
	int iResult = 1;

	if ( argc < 4 ) {
		printf(
			"usage: stream <rsa|p256|p384|ed25519> "
			"<certificate.der> <private.der> [port]\n"
		);
		return 0;
	}
	if ( argc >= 5 ) {
		char* pEnd = NULL;

		iPort = strtoul(argv[4], &pEnd, 10);
		if ( (pEnd == argv[4]) || (*pEnd != '\0') ||
			(iPort == 0) || (iPort > 65535u) ) {
			fprintf(stderr, "invalid port\n");
			goto Cleanup;
		}
	}
	if ( !exampleTlsReadFile(
		argv[2],
		&pCertificateData,
		&iCertificateSize
	) || !exampleTlsReadFile(
		argv[3],
		&pPrivateData,
		&iPrivateSize
	) ) {
		fprintf(stderr, "failed to read DER identity\n");
		goto Cleanup;
	}
	Certificate = (xbytesview) {
		pCertificateData,
		iCertificateSize
	};
	pIdentity = exampleTlsIdentity(
		argv[1],
		&Certificate,
		1u,
		(xbytesview) { pPrivateData, iPrivateSize }
	);
	if ( pIdentity == NULL ) {
		exampleTlsStreamError("failed to create TLS identity");
		goto Cleanup;
	}

	memset(&Example, 0, sizeof(Example));
	memset(&ListenerEvents, 0, sizeof(ListenerEvents));
	xrtTlsServerConfigInit(&Example.ServerConfig);
	Example.ServerConfig.Identity = pIdentity;
	Example.ServerConfig.Protocols = Protocols;
	Example.ServerConfig.ProtocolCount = 1u;
	Example.ServerConfig.RequireProtocol = true;
	xrtTlsStreamConfigInit(&Example.StreamConfig);
	Example.StreamEvents.Open = exampleTlsStreamOpen;
	Example.StreamEvents.Read = exampleTlsStreamRead;
	Example.StreamEvents.Writable = exampleTlsStreamWritable;
	Example.StreamEvents.Close = exampleTlsStreamClose;
	ListenerEvents.Accept = exampleTlsStreamAccept;
	ListenerEvents.Close = exampleTlsStreamListenerClose;

	xrtNetEngineConfigInit(&EngineConfig);
	pEngine = xrtNetEngineCreate(&EngineConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		exampleTlsStreamError("failed to start network engine");
		goto Cleanup;
	}
	xrtNetListenConfigInit(&ListenConfig);
	if ( !xrtNetAddrAny(
		&ListenConfig.Address,
		XNET_FAMILY_IPV4,
		(uint16)iPort
	) ) {
		exampleTlsStreamError("failed to build listen address");
		goto Cleanup;
	}
	pListener = xrtNetListen(
		pEngine,
		&ListenConfig,
		&ListenerEvents,
		NULL,
		&Example
	);
	if ( (pListener == NULL) ||
		!xrtNetListenerLocal(pListener, &Address) ) {
		exampleTlsStreamError("failed to listen");
		goto Cleanup;
	}
	printf(
		"TLS echo listening on 0.0.0.0:%u; press Enter to stop\n",
		(unsigned)Address.Port
	);
	(void)getchar();
	if ( !xrtNetListenerClose(pListener) ) {
		exampleTlsStreamError("failed to close listener");
		goto Cleanup;
	}
	while ( (xrtAtomic32Load(
			&Example.ListenerClosed,
			XMEMORY_ACQUIRE
		) == 0) || (xrtAtomic32Load(
			&Example.Connections,
			XMEMORY_ACQUIRE
		) != 0) ) {
		xrtThreadYield();
	}
	iResult = 0;

Cleanup:
	xrtNetListenerDestroy(pListener);
	if ( (pEngine != NULL) && !xrtNetEngineDestroy(pEngine) ) {
		exampleTlsStreamError("failed to destroy network engine");
		iResult = 1;
	}
	xrtTlsIdentityRelease(pIdentity);
	free(pPrivateData);
	free(pCertificateData);
	return iResult;
}

$ gcc -O1 -DXRT_MODULE_ALL -I single -include xrt.h impl.c examples/tls/stream/main.c -lws2_32 -liphlpapiusage: stream <rsa|p256|p384|ed25519> ...

刚才发生了什么。 ① Echo 主循环是硬背压的范本：Available → Buffer → Front 取 Span → Send(部分受理 iWritten) → Consume(iWritten) → AGAIN 即返回——Read 与 Writable 都进这个无重复路径，慢客户端永远拖不爆内存。② Accept 接线：TCP Accept 回调里 StreamAccept 一步建组合对象——本例随即 Destroy（演示所有权：Accept 成功即拿到引用，事件表已接管后续；真实服务保持连接自然运行到关闭）。③ Open 里的传输查询：StreamTransport 借出 TCP 流查 Remote 地址打印"TLS client open: 地址"——TLS 层与传输层的桥就这么一座。④ 生命周期汇合：Close 回调递减 Connections、ListenerClose 置位——主线程等两者归零再释放共享配置；这就是"N 连接共享一份装配"的关闭侧编排。⑤ 对照第 67 章的 TCP 回显并排读：事件模型、Span 消费、短写续传逐条同构——多出来的只有握手与两级关闭。

第二个完整程序：Stream 层全特性巡检

第二个程序来自 examples/tls/stream_tour/main.c——五行为你覆盖 Stream 层的进阶面：

examples/tls/stream_tour/main.c源码 ↗



























 */

#include "../embedded_identity.h"

#include <stdio.h>
#include <string.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 EXAMPLE_DEADLINE_MS	INT64_C(5000)

/* 客户端上下文：Open/Read 回调与主线程交接。 */
typedef struct example_tls_upgrade {
	volatile bool bOpen;
	volatile bool bClosed;
	volatile size_t Received;
	size_t iPullup;
	size_t iRead;
	size_t iConsume;
	char ReadOut[8];
	bool bSessionOk;
	bool bDataOk;
	bool bReadMore;
	bool bEventsSwapped;
	size_t iBound;
} example_tls_upgrade;

/* Worker 任务上下文：TCP 连接后升级为 TLS。 */
typedef struct example_upgrade_task {
	xnetstream* pTcp;
	xtlssession* pSession;  /* 仅 Attach 形态非空 */
	const xtlsclientconfig* pClient;
	const xtlsstreamconfig* pStream;
	const xtlsstreamevents* pEvents;
	ptr pData;
	xtlsstream* pTls;
	volatile bool bDone;
	bool bOk;
} example_upgrade_task;

/* 客户端事件集与热替换副本：文件级存储保证回调期间存活。 */
static xtlsstreamevents g_ClientEvents;
static xtlsstreamevents g_SwappedEvents;

static xtlsverifydecision exampleAcceptAll(
	const xtlspeer* pPeer, ptr pContext)
{
	(void)pPeer;
	(void)pContext;
	return XTLS_VERIFY_ACCEPT;
}

static bool exampleSpinUntil(volatile bool* pFlag)
{
	double iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);

	while ( !*pFlag ) {
		if ( exampleTimerExpired(iDeadline) ) {
			return false;
		}
		xrtThreadYield();
	}
	return true;
}

/* ---- HTTP over TLS 段：ParseTls 双向解析。 ---- */
static volatile bool g_bHttpReqParsed;
static volatile bool g_bHttpRspParsed;
static volatile bool g_bHttpMode;

/* Worker 任务：热替换事件集 / 发送请求。 */
typedef struct example_http_task {
	xtlsstream* pStream;
	volatile bool bDone;
	bool bOk;
} example_http_task;

/* 请求与应答全文用 sizeof 定长——少一字节 Header 即不完整。 */
static const char arrRequest[] = "GET /x HTTP/1.1\r\nHost: h\r\n\r\n";
static const char arrResponse[] =
	"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nhi";

static void exampleHttpSendTask(xnetworker* pWorker, ptr pUserData)
{
	example_http_task* pTask = (example_http_task*)pUserData;
	size_t iWritten = 0;

	(void)pWorker;
	pTask->bOk = (xrtTlsStreamSend(pTask->pStream, arrRequest,
		sizeof(arrRequest) - 1u,
		&iWritten) == XTLS_OK) &&
		(iWritten == sizeof(arrRequest) - 1u);
	pTask->bDone = true;
}

/* Worker 任务：StreamClient（pSession == NULL）或 StreamAttach。 */
static void exampleUpgradeTask(xnetworker* pWorker, ptr pUserData)
{
	example_upgrade_task* pTask = (example_upgrade_task*)pUserData;

	(void)pWorker;
	if ( pTask->pSession != NULL ) {
		pTask->bOk = xrtTlsStreamAttach(pTask->pTcp,
			pTask->pSession, pTask->pStream, pTask->pEvents,
			pTask->pData, &pTask->pTls);
	}
	else {
		pTask->bOk = xrtTlsStreamClient(pTask->pTcp,
			pTask->pClient, pTask->pStream, pTask->pEvents,
			pTask->pData, &pTask->pTls);
	}
	pTask->bDone = true;
}

/* 服务端 Read 回调（Worker）：从 TLS 明文块链解析请求并应答。 */
static void exampleHttpServerRead(xtlsstream* pStream,
	const xnetbuf* pBuffer, ptr pData)
{
	xhttp1head Head;
	xhttpfield Fields[8];
	xhttp1limits Limits;
	xhttp1errorinfo Error;
	size_t iWritten = 0;
	xnetspan Span;

	(void)pBuffer;
	(void)pData;
	
	 * 以 "GET" 起头的请求前缀。 */
	while ( xrtTlsStreamAvailable(pStream) >= 3u ) {
		if ( !xrtTlsStreamPullup(pStream, 3u, &Span) ||
			(Span.Size < 3u) ) {
			return;
		}
		if ( memcmp(Span.Data, "GET", 3u) == 0 ) {
			break;
		}
		if ( !xrtTlsStreamConsume(pStream, 1u) ) {
			return;
		}
	}
	xrtHttp1LimitsInit(&Limits);
	xrtHttp1HeadInit(&Head, Fields, 8);
	if ( (xrtHttp1RequestParseTls(pStream, &Head, &Limits,
			&Error) != XHTTP1_READY) ||
		(Head.Method.Size != 3u) ||
		(memcmp(Head.Method.Data, "GET", 3u) != 0) ||
		(Head.Target.Size != 2u) ||
		(memcmp(Head.Target.Data, "/x", 2u) != 0) ||
		(xrtTlsStreamSend(pStream, arrResponse,
			sizeof(arrResponse) - 1u,
			&iWritten) != XTLS_OK) ||
		(iWritten != sizeof(arrResponse) - 1u) ) {
		return;
	}
	g_bHttpReqParsed = true;
}

/* 客户端 Read 回调（Worker）：从明文块链解析响应。 */
static void exampleHttpClientRead(xtlsstream* pStream,
	const xnetbuf* pBuffer, ptr pData)
{
	xhttp1head Head;
	xhttpfield Fields[8];
	xhttp1limits Limits;
	xhttp1errorinfo Error;

	(void)pBuffer;
	(void)pData;
	xrtHttp1LimitsInit(&Limits);
	xrtHttp1HeadInit(&Head, Fields, 8);
	if ( (xrtHttp1ResponseParseTls(pStream, &Head, &Limits,
			&Error) == XHTTP1_READY) &&
		(Head.Status == 200u) &&
		(Head.ContentLength == 2u) ) {
		g_bHttpRspParsed = true;
	}
}

/* 服务端回显：借用明文前缀逐块回送，短写等 Writable 再续。 */
static void exampleServerRead(xtlsstream* pStream,
	const xnetbuf* pBuffer, ptr pData)
{
	(void)pBuffer;
	(void)pData;
	if ( g_bHttpMode ) {
		exampleHttpServerRead(pStream, NULL, NULL);
		return;
	}
	while ( xrtTlsStreamAvailable(pStream) != 0 ) {
		const xnetbuf* pPlain = xrtTlsStreamBuffer(pStream);
		xnetspan Span;
		size_t iWritten = 0;

		if ( (pPlain == NULL) ||
			!xrtNetBufFront(pPlain, &Span) ) {
			(void)xrtTlsStreamAbort(pStream);
			return;
		}
		if ( xrtTlsStreamSend(pStream, Span.Data, Span.Size,
			&iWritten) != XTLS_OK ) {
			(void)xrtTlsStreamAbort(pStream);
			return;
		}
		if ( (iWritten != 0) &&
			!xrtTlsStreamConsume(pStream, iWritten) ) {
			(void)xrtTlsStreamAbort(pStream);
			return;
		}
	}
}

static void exampleUpgradeOpen(xtlsstream* pStream, ptr pData)
{
	example_tls_upgrade* pClient = (example_tls_upgrade*)pData;
	xnetspan Vec[2];
	size_t iWritten = 0;

	/* Open 回调（Worker 上）：聚集发送一条 9 字节明文。 */
	Vec[0].Data = (cbytes)"vec-";
	Vec[0].Size = 4;
	Vec[1].Data = (cbytes)"gather";
	Vec[1].Size = 5;
	(void)xrtTlsStreamSendVec(pStream, Vec, 2, &iWritten);
	/* SendBound 是 Worker 专用：明文尺寸 → 密文上界。 */
	(void)xrtTlsStreamSendBound(pStream, 64u, &pClient->iBound);
	
	 * （先复制处理器再替换，语义等价但演示了热替换入口）。 */
	g_SwappedEvents = g_ClientEvents;
	pClient->bEventsSwapped = xrtTlsStreamSetEvents(pStream,
		&g_SwappedEvents, pClient);
	pClient->bSessionOk = xrtTlsStreamSession(pStream) != NULL;
	pClient->bDataOk = xrtTlsStreamData(pStream) == pClient;
	pClient->bOpen = true;
}

static void exampleUpgradeRead(xtlsstream* pStream,
	const xnetbuf* pBuffer, ptr pData)
{
	if ( g_bHttpMode ) {
		exampleHttpClientRead(pStream, NULL, NULL);
		return;
	}

	example_tls_upgrade* pClient = (example_tls_upgrade*)pData;
	size_t iAvail = xrtTlsStreamAvailable(pStream);
	size_t iWant = iAvail < 9u ? iAvail : 9u;
	xnetspan Span;

	(void)pBuffer;
	
	 * Pullup 按当前可用前缀请求连续化，而不是按期望总长。 */
	if ( (iWant != 0u) &&
		xrtTlsStreamPullup(pStream, iWant, &Span) &&
		(Span.Size == iWant) ) {
		pClient->iPullup = iWant;  /* 当前已连续化的前缀长度 */
	}
	/* 异步在途观测：Worker 内快照（主线程同样可读）。 */
	(void)xrtTlsStreamAsyncBytes(pStream);
	(void)xrtTlsStreamAsyncCount(pStream);
	if ( iAvail < 9u ) {
		/* 消息未到齐：ReadMore 在保留已到前缀的前提下请求继续累积。 */
		pClient->bReadMore = xrtTlsStreamReadMore(pStream);
		return;
	}
	
	 * 剩余 5 字节用 Consume 精确丢弃——Read 本身就是消费式复制。 */
	pClient->bReadMore = xrtTlsStreamReadMore(pStream);
	pClient->iRead = 0;
	(void)xrtTlsStreamRead(pStream, pClient->ReadOut, 4u,
		&pClient->iRead);
	if ( xrtTlsStreamConsume(pStream, 5u) ) {
		pClient->iConsume = 5u;
	}
	pClient->Received = pClient->Received + 9u;
}

static void exampleUpgradeClose(xtlsstream* pStream,
	xnetresult Result, const xerror* pError, ptr pData)
{
	example_tls_upgrade* pClient = (example_tls_upgrade*)pData;

	(void)pStream;
	(void)Result;
	(void)pError;
	pClient->bClosed = true;
}

/* 等一条 TCP Stream 进入 OPEN。 */
static bool exampleWaitTcpOpen(xnetstream* pTcp)
{
	double iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);

	while ( xrtNetStreamState(pTcp) != XNET_STREAM_OPEN ) {
		if ( exampleTimerExpired(iDeadline) ) {
			return false;
		}
		xrtThreadYield();
	}
	return true;
}

int main(void)
{
	xnetengineconfig EngineConfig;
	xnetresolverconfig ResolverConfig;
	xtlslistenerconfig ListenerConfig;
	xtlsserverconfig ServerConfig;
	xtlsclientconfig ClientConfig;
	xtlsverifierconfig VerifierConfig;
	xtlsstreamconfig StreamConfig;

	xtlsstreamevents ServerEvents;
	xnetengine* pEngine = NULL;
	xnetresolver* pResolver = NULL;
	xtlsverifier* pVerifier = NULL;
	xtlsidentity* pIdentity = NULL;
	xtlslistener* pListener = NULL;
	xnetstream* pTcpA = NULL;
	xnetstream* pTcpB = NULL;
	xtlssession* pSessionB = NULL;
	xtlsstream* pClientA = NULL;
	xtlsstream* pClientB = NULL;
	xtlsstream* pServerA = NULL;
	xtlsstream* pServerB = NULL;
	example_tls_upgrade ClientA;
	example_tls_upgrade ClientB;
	example_upgrade_task TaskA;
	example_upgrade_task TaskB;
	xnetpost Post;
	xfuture* pSendFuture = NULL;
	xnetaddr Address;
	xnetspan AsyncVec[2];
	size_t iBound = 0;
	double iDeadline;
	int iResult = 1;

	memset(&ClientA, 0, sizeof(ClientA));
	memset(&ClientB, 0, sizeof(ClientB));

	/* 服务端：回显 Listener（Read 事件原样回送）。 */
	pIdentity = exampleEmbeddedIdentity();
	if ( pIdentity == NULL ) {
		fprintf(stderr, "failed to create embedded identity\n");
		goto Cleanup;
	}
	xrtTlsServerConfigInit(&ServerConfig);
	ServerConfig.Identity = pIdentity;
	xrtTlsListenerConfigInit(&ListenerConfig);
	ListenerConfig.Tls = ServerConfig;
	(void)xrtNetAddrLoopback(&ListenerConfig.Listen.Address,
		XNET_FAMILY_IPV4, 0);
	memset(&ServerEvents, 0, sizeof(ServerEvents));
	ServerEvents.Read = exampleServerRead;

	/* 客户端：接受一切 + 显式 VerifyName（直连不走 Dial 名字推导）。 */
	xrtTlsVerifierConfigInit(&VerifierConfig);
	VerifierConfig.Verify = exampleAcceptAll;
	pVerifier = xrtTlsVerifierCreate(&VerifierConfig);
	if ( pVerifier == NULL ) {
		fprintf(stderr, "failed to create verifier\n");
		goto Cleanup;
	}
	xrtTlsClientConfigInit(&ClientConfig);
	ClientConfig.Verifier = pVerifier;
	ClientConfig.VerifyName = XRT_STR_LITERAL("localhost");
	xrtTlsStreamConfigInit(&StreamConfig);
	memset(&g_ClientEvents, 0, sizeof(g_ClientEvents));
	g_ClientEvents.Open = exampleUpgradeOpen;
	g_ClientEvents.Read = exampleUpgradeRead;
	g_ClientEvents.Close = exampleUpgradeClose;

	xrtNetEngineConfigInit(&EngineConfig);
	EngineConfig.Workers = 1;
	xrtNetResolverConfigInit(&ResolverConfig);
	pEngine = xrtNetEngineCreate(&EngineConfig);
	pResolver = xrtNetResolverCreate(&ResolverConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ||
		(pResolver == NULL) ) {
		fprintf(stderr, "failed to start engine/resolver\n");
		goto Cleanup;
	}
	/* StreamEvents 让每条被接受的连接自动回显（队列中同样生效）。 */
	pListener = xrtTlsListenerStart(pEngine, &ListenerConfig, NULL,
		&ServerEvents, NULL);
	if ( (pListener == NULL) ||
		!xrtTlsListenerLocal(pListener, &Address) ) {
		iResult = 2;
		goto Cleanup;
	}

	/* 形态一（STARTTLS）：裸 TCP → Worker 上 StreamClient 升级。 */
	pTcpA = xrtNetStreamConnect(pEngine, &Address, 0, NULL,
		NULL, NULL);
	if ( (pTcpA == NULL) || !exampleWaitTcpOpen(pTcpA) ) {
		iResult = 3;
		goto Cleanup;
	}
	memset(&TaskA, 0, sizeof(TaskA));
	TaskA.pTcp = pTcpA;
	TaskA.pClient = &ClientConfig;
	TaskA.pStream = &StreamConfig;
	TaskA.pEvents = &g_ClientEvents;
	TaskA.pData = &ClientA;
	if ( !xrtNetPostInit(&Post) || !xrtNetPost(
		xrtNetStreamWorker(pTcpA), &Post, exampleUpgradeTask,
		&TaskA) || !exampleSpinUntil(&TaskA.bDone) ||
		!TaskA.bOk ) {
		iResult = 3;
		goto Cleanup;
	}
	pClientA = TaskA.pTls;

	/* 形态二（Attach）：显式会话 + 裸 TCP → Worker 上 Attach 组合。 */
	pSessionB = xrtTlsClientCreate(&ClientConfig, NULL);
	pTcpB = xrtNetStreamConnect(pEngine, &Address, 0, NULL,
		NULL, NULL);
	if ( (pSessionB == NULL) || (pTcpB == NULL) ||
		!exampleWaitTcpOpen(pTcpB) ) {
		iResult = 4;
		goto Cleanup;
	}
	memset(&TaskB, 0, sizeof(TaskB));
	TaskB.pTcp = pTcpB;
	TaskB.pSession = pSessionB;
	TaskB.pStream = &StreamConfig;
	TaskB.pEvents = &g_ClientEvents;
	TaskB.pData = &ClientB;
	if ( !xrtNetPostInit(&Post) || !xrtNetPost(
		xrtNetStreamWorker(pTcpB), &Post, exampleUpgradeTask,
		&TaskB) || !exampleSpinUntil(&TaskB.bDone) ||
		!TaskB.bOk ) {
		iResult = 4;
		goto Cleanup;
	}
	pClientB = TaskB.pTls;
	/* Attach 成功即接管调用方的 Session 引用；这里不能再 Destroy。 */
	pSessionB = NULL;

	/* 两个客户端都握手完成。 */
	if ( !exampleSpinUntil(&ClientA.bOpen) ||
		!exampleSpinUntil(&ClientB.bOpen) ) {
		iResult = 5;
		goto Cleanup;
	}
	printf("stream-tour: two clients upgraded, echo verified\n");

	/* 取出两条服务端流：队列中的连接被 Accept 消费后才开始驱动回显。 */
	{
		double iEnd = exampleTimerLimit(EXAMPLE_DEADLINE_MS);

		while ( (pServerA == NULL) || (pServerB == NULL) ) {
			xtlsstream* pOne = xrtTlsListenerAccept(pListener);

			if ( pOne != NULL ) {
				if ( pServerA == NULL ) {
					pServerA = pOne;
				}
				else {
					pServerB = pOne;
				}
				iEnd = exampleTimerLimit(
					EXAMPLE_DEADLINE_MS);
				continue;
			}
			if ( exampleTimerExpired(iEnd) ) {
				iResult = 6;
				goto Cleanup;
			}
			xrtThreadYield();
		}
	}

	/* 回显到达：Read 回调里完成 Pullup/Read/Consume 核对。 */
	iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);
	while ( ClientA.Received < 9u ) {
		if ( exampleTimerExpired(iDeadline) ) {
			iResult = 6;
			goto Cleanup;
		}
		xrtThreadYield();
	}
	if ( (ClientA.iPullup != 9u) || (ClientA.iRead != 4u) ||
		(ClientA.iConsume != 5u) ||
		(memcmp(ClientA.ReadOut, "vec-", 4) != 0) ||
		!ClientA.bSessionOk || !ClientA.bDataOk ||
		!ClientA.bReadMore || !ClientA.bEventsSwapped ||
		(ClientA.iBound < 64u) ) {
		iResult = 6;
		goto Cleanup;
	}
	printf("client(starttls): vec=9 bytes read=4 consume=%zu",
		ClientA.iConsume);
	printf(" session+data=%s\n",
		(ClientA.bSessionOk && ClientA.bDataOk) ? "ok" : "fail");

	/* Attach 客户端：SendVecAsync（Future 式聚集发送）。 */
	AsyncVec[0].Data = (cbytes)"asy-";
	AsyncVec[0].Size = 4;
	AsyncVec[1].Data = (cbytes)"ncvec";
	AsyncVec[1].Size = 5;
	pSendFuture = xrtTlsStreamSendVecAsync(pClientB, AsyncVec, 2);
	if ( (pSendFuture == NULL) ||
		(xrtFutureWaitFor(pSendFuture, EXAMPLE_DEADLINE_MS) !=
			XWAIT_OK) ||
		(xrtFutureState(pSendFuture) != XFUTURE_RESOLVED) ) {
		iResult = 7;
		goto Cleanup;
	}
	iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);
	while ( ClientB.Received < 9u ) {
		if ( exampleTimerExpired(iDeadline) ) {
			iResult = 7;
			goto Cleanup;
		}
		xrtThreadYield();
	}
	printf("client(attach): async-vec=ok future-written=9");

	/* Pending：发送队列排空。 */
	iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);
	while ( xrtTlsStreamPending(pClientB) != 0u ) {
		if ( exampleTimerExpired(iDeadline) ) {
			iResult = 8;
			goto Cleanup;
		}
		xrtThreadYield();
	}
	printf(" pending=0\n");

	/* SendBound 的结果已在 Open 回调（Worker 内）算好带回。 */
	iBound = ClientA.iBound;
	printf("bounds: 64 -> %zu ciphertext, error=%s\n", iBound,
		xrtTlsStreamError(pClientA) == NULL ? "(none)" : "err");
	(void)xrtTlsStreamAsyncBytes(pClientA);
	(void)xrtTlsStreamAsyncCount(pClientA);
	
	 * 双方 Read 回调内用 ParseTls 从明文块链解析。 ---- */
	g_bHttpMode = true;
	{
		example_http_task Send;

		memset(&Send, 0, sizeof(Send));
		Send.pStream = pClientB;
		if ( !xrtNetPost(xrtNetStreamWorker(pTcpB), &Post,
				exampleHttpSendTask, &Send) ||
			!exampleSpinUntil(&Send.bDone) || !Send.bOk ) {
			iResult = 9;
			goto Cleanup;
		}
	}
	iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);
	while ( !g_bHttpReqParsed || !g_bHttpRspParsed ) {
		if ( exampleTimerExpired(iDeadline) ) {
			iResult = 9;
			goto Cleanup;
		}
		xrtThreadYield();
	}
	printf("http1-tls: request+response parsed on stream ok\n");
	iResult = 0;

Cleanup:
	xrtFutureDestroy(pSendFuture);
	
	 * 共享截止时间等终态，超时者 Abort 兜底。 */
	{
		xtlsstream* Streams[4];
		double iEnd = exampleTimerLimit(EXAMPLE_DEADLINE_MS);
		bool bSettled;

		Streams[0] = pClientA;
		Streams[1] = pClientB;
		Streams[2] = pServerA;
		Streams[3] = pServerB;
		for ( size_t i = 0; i < 4; ++i ) {
			if ( Streams[i] != NULL ) {
				(void)xrtTlsStreamClose(Streams[i]);
			}
		}
		do {
			bSettled = true;
			for ( size_t i = 0; i < 4; ++i ) {
				xtlsstreamstate State = Streams[i] != NULL ?
					xrtTlsStreamState(Streams[i]) :
					XTLS_STREAM_CLOSED;

				if ( (State != XTLS_STREAM_CLOSED) &&
					(State != XTLS_STREAM_FAILED) ) {
					bSettled = false;
				}
			}
			if ( bSettled || exampleTimerExpired(iEnd) ) {
				break;
			}
			xrtThreadYield();
		} while ( 1 );
		for ( size_t i = 0; i < 4; ++i ) {
			if ( Streams[i] != NULL ) {
				xtlsstreamstate State =
					xrtTlsStreamState(Streams[i]);

				if ( (State != XTLS_STREAM_CLOSED) &&
					(State != XTLS_STREAM_FAILED) ) {
					(void)xrtTlsStreamAbort(Streams[i]);
				}
			}
		}
		(void)exampleSpinUntil(&ClientA.bClosed);
		(void)exampleSpinUntil(&ClientB.bClosed);
	}
	xrtTlsStreamDestroy(pClientA);
	xrtTlsStreamDestroy(pClientB);
	xrtTlsStreamDestroy(pServerA);
	xrtTlsStreamDestroy(pServerB);
	xrtTlsSessionDestroy(pSessionB);
	if ( pTcpA != NULL ) {
		(void)xrtNetStreamAbort(pTcpA);
	}
	if ( pTcpB != NULL ) {
		(void)xrtNetStreamAbort(pTcpB);
	}
	xrtNetStreamDestroy(pTcpA);
	xrtNetStreamDestroy(pTcpB);
	if ( pListener != NULL ) {
		double iEnd = exampleTimerLimit(EXAMPLE_DEADLINE_MS);

		(void)xrtTlsListenerClose(pListener);
		while ( xrtTlsListenerState(pListener) !=
			XTLS_LISTENER_CLOSED ) {
			if ( exampleTimerExpired(iEnd) ) {
				break;
			}
			xrtThreadYield();
		}
	}
	xrtTlsListenerDestroy(pListener);
	xrtNetResolverDestroy(pResolver);
	if ( (pEngine != NULL) && !xrtNetEngineDestroy(pEngine) &&
		(iResult == 0) ) {
		iResult = 10;
	}
	xrtTlsVerifierRelease(pVerifier);
	xrtTlsIdentityRelease(pIdentity);
	return iResult;
}

$ gcc -O1 -DXRT_MODULE_ALL -I single -include xrt.h impl.c examples/tls/stream_tour/main.c -lws2_32 -liphlpapistream-tour: two clients upgraded, echo verifiedclient(starttls): vec=9 bytes read=4 consume=5 session+data=okclient(attach): async-vec=ok future-written=9 pending=0bounds: 64 -> 86 ciphertext, error=(none)http1-tls: request+response parsed on stream ok

刚才发生了什么。 ① 第 1 行两条客户端接入路径同检：starttls（明文连接上协商升级——第 87 章练习的官方答案）与 attach（TCP 流直挂）——加上服务端 Accept，Stream 层的三种建立方式全集。② 第 2 行演示 Vec 批量发送（9 字节）与 Span 消费（读 4 消费 5——消费可以超前于读取边界，水位语义）。③ 第 3 行走异步 Vec 写入：Future 完成后 Pending 归零——第 68 章发送五档的 Vec/异步档在 TLS 流上的样子。④ 第 4 行测量密文边界：64 字节明文 → 86 字节密文（16 字节 AEAD 标签 + 记录头开销）——给容量规划一个真实锚点。⑤ 第 5 行在流上跑完整 HTTP/1 请求响应解析——把第 84～87 章的原语串成一条链的自检样本，也是卷九的预告。

87.4契约

建立：服务端 StreamAccept（TCP Accept 回调内一步）；客户端经 TlsDial/DialAsync 内联 attach；两侧共享 xtlsstream 全部应用接口。
事件：Open/Read/Writable/Close 与 TCP 流同构；回调在 Engine Worker 执行；Close 是唯一终态通知（结果 + 结构化错误）。
硬背压：明文消费停滞即接收停滞——Span 逐块回送 + 短写等 Writable 是标准回显形态；内存与最慢客户端解耦。
传输桥：StreamTransport 借出底层 TCP 流（只读查询：地址/状态）；不可缓存跨回调使用。
两级关闭：Close 走 close_notify 排空；Abort 立即中止；单连接等 CLOSED/FAILED 再 Destroy；服务级等"全部连接 + 监听器"归零再释放共享配置。
配置分离：xtlsserverconfig（协议面）与 xtlsstreamconfig（流面）各自独立；配置对象共享引用计数。
握手自动化：Feed/Drive 循环由组合层执行——应用从不见密文（stream_tour 的边界测量属诊断工具用法）。
裁剪：XRT_MODULE_TLS_STREAM 独立（依赖 NET）——只要会话机不要传输组合时可裁掉。

常见错误

坑 1：Read 回调里把明文复制完再慢慢处理

症状：高吞吐下内存尖峰——每条连接的 Read 回调都把接收区整份复制进自己的队列，背压机制被架空。

原因：Stream 层的背压建立在"消费明文（Consume）"上；复制不清费等于没消费，接收侧继续推进，你的队列成了无界缓冲。

错误写法

static void onRead(xtlsstream* pStream, const xnetbuf* pBuffer, ptr pData) {
	queue_push(g_MyQueue, pBuffer);   /* 复制走但不 Consume：背压失效 */
}

正确写法

static void onRead(xtlsstream* pStream, const xnetbuf* pBuffer, ptr pData) {
	/* 能处理多少消费多少；处理不了的留给下一次 Read/Writable */
	while ( xrtTlsStreamAvailable(pStream) != 0 ) {
		/* ... Front 取 Span 处理 + Consume ... */
		if ( 下游满了 ) { return; }   /* 停消费＝停接收＝背压生效 */
	}
}

坑 2：监听器关了就立刻释放共享配置

症状：最后的连接回调里访问已释放的身份/配置——崩溃在最后一个客户端断开时。

原因：共享配置的引用计数在每条连接上——Listener 关闭不等于连接关闭；释放必须等两个计数都归零。

错误写法

listenerClose(...) {
	xrtTlsIdentityRelease(g_Identity);  /* 还有活跃连接在用 */
}

正确写法

/* 原子计数汇合：Connections==0 && ListenerClosed 才统一释放 */
if ( (xrtAtomic32Load(&Connections, XMEMORY_ACQUIRE) == 0) &&
		(xrtAtomic32Load(&ListenerClosed, XMEMORY_ACQUIRE) != 0) ) {
	release_shared_assembly();
}
/* Close 与 ListenerClose 两个回调都做这个检查 */

坑 3：把 StreamTransport 借出的 TCP 流存起来长期用

症状：偶发错乱——缓存的 TCP 流指针在组合对象内部状态推进后失配。

原因：StreamTransport 是"此刻的传输桥"诊断/查询视图，不是长期句柄。要用传输能力（查地址、状态）就在需要的时刻现取现用。

错误写法

g_SavedTcp = xrtTlsStreamTransport(pStream);   /* 存起来 */
/* 若干回调之后 */
send_on_tcp(g_SavedTcp, ...);   /* 绕过 TLS 层直写传输：状态失配 */

正确写法

/* 需要时现查（如 Open 里记日志）；发送永远走 TLS 层的 Send */
if ( xrtNetStreamRemote(xrtTlsStreamTransport(pStream), &Remote) ) {
	log_client(Remote);
}
xrtTlsStreamSend(pStream, Data, Size, &Written);  /* 明文走正门 */

编程练习

基础：openssl 联调

生成自签证书跑通 stream 服务，用 openssl s_client -connect localhost:8443 -alpn http/1.1 连接，输入文本验证回显；观察服务端打印的客户端地址。验收标准：回显逐字节正确；断开（Ctrl-C 或 Q）后服务端 Close 回调触发、连接计数归零、进程不退（继续服务下一个连接）。

进阶：明文代理（解密侧观察）

在 Echo 之前插一层"审计"：Read 回调里统计每条连接收到的明文字节数与 Span 数，Close 时输出汇总（总字节/平均 Span 大小/连接时长）。验收标准：并发 4 连接各自独立统计；数据与 openssl 侧发送量一致；统计不复制明文（零拷贝路径上完成）。

挑战：StartTLS 升级网关

实现"先明文握手命令再升级"的网关：TCP Accept 后不立即 TLS，先按行读命令（第 72 章行分帧），收到 STARTTLS\r\n 后调用 StreamAttach 升级为 TLS（其余命令按明文模式回显）。提示：升级前后的同一连接对象——查 stream_tour 的 starttls 客户端路径作对照。验收标准：明文阶段命令正常回显；升级后明文命令被拒绝；openssl 经 starttls 提示可完成升级后加密会话；连接计数与两种阶段的资源都正确清理。

小结

知识点	速查
组合对象	服务端 `StreamAccept`（Accept 回调内一步）/ 客户端 `TlsDial` 内联 attach
事件模型	Open/Read/Writable/Close 与 TCP 流同构；Worker 线程执行
硬背压	停消费＝停接收；Span 回送 + 短写等 Writable；内存与最慢客户端解耦
应用接口	Send（部分受理）/ Available/Buffer/Front/Consume / Close / Abort / State
传输桥	`StreamTransport` 借出 TCP 流——现查现用不缓存；发送走 TLS 层正门
配置分离	xtlsserverconfig（协议面）/ xtlsstreamconfig（流面）；共享引用计数
服务级生命周期	全部连接 + 监听器归零后统一释放共享装配（原子计数汇合）
单连接生命周期	Close/Abort → 等 CLOSED/FAILED → Destroy（第 82 章规则）
握手自动化	Feed/Drive 由组合层执行，应用从不见密文
裁剪	`XRT_MODULE_TLS_STREAM` 独立于会话机；stream_tour 是全特性自检

上一章第 86 章 TLS（中）：服务端下一章第 88 章会话恢复：票据、PSK 与对象契约

📖 本章涉及的 API 参考手册：tls（201 个条目）tls_stream（67 个条目）net（247 个条目）

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