Chapter 87 TLS Streams: The Composition Layer over TCP

Chapter Orientation

Chapter 86's session machine touches no network; Chapter 82's client dial already quietly used the Stream layer. This chapter finishes the Stream layer (xtlsstream) head-on: the server shape — xrtTlsStreamAccept in the TCP Accept callback builds the "TCP stream + TLS session machine" composite in one step, after which the business sees only plaintext read/write; the event model — the four callbacks Open/Read/Writable/Close (the same set as Chapter 82's client); backpressure — plaintext receive Available + short-write waiting on Writable; a slow client never balloons server memory; the lifecycle — shared configuration reference-counted, released together after the listener and every connection close. This is another redemption of the library-wide "skeleton composition" pattern: Engine (Chapter 64) + Listener/Stream (Chapter 67) + TLS session machine (Chapter 86) assembled into a complete service openssl s_client can connect to directly.

87.1引入

Turning Chapter 86's echo sparring into a real service requires only "transport wiring": Accept a TCP stream → build the composite → the handshake runs automatically → event callbacks deliver plaintext. It sounds like a thin layer, but every spot is a trap: who feeds the handshake bytes? Where does plaintext consumption pause when write backpressure jams? How do connection close and listener close coordinate without leaks? The Stream layer standardizes this wiring — what you write is the business logic inside the Echo callback; lifecycle and backpressure are managed by the composite.

The service shape is worth seeing: one process, one Engine, one Listener, N TLS connections sharing the same server configuration (identity/ALPN) and event table — reference-counted configuration makes "assemble once, reuse across N connections" zero-copy. This lines up with Chapter 83's identity sharing and Chapter 85's verifier sharing: all of TLS's heavy objects are immutable shared values.

87.2概念

Building the composite: Accept and Dial sides

Server: xrtTlsStreamAccept(TCP流, 服务端配置, 流配置, 事件表, 用户数据, &流) — inside the TCP Accept callback, building the composite in one step; the handshake runs automatically, and the Open callback fires on success.
Client: xrtTlsDial/DialAsync (Chapter 82) internally is exactly "TCP dial + StreamAttach in the client direction" — the client has no separate Attach example because dialing already inlines it.

Both sides share the xtlsstream type and every application interface — Send/Available/Buffer/Consume/Close/Abort/State (Chapter 82's cheat sheet). They differ only in how they are built and their direction.

The event model and hard backpressure

The four callbacks are fully isomorphic to the TCP stream's (Chapter 67): Open (handshake complete/READY), Read (plaintext arrived), Writable (send budget recovered), Close (the sole final state: xnetresult + structured error). The hard-backpressure mechanism: the plaintext receive area has a watermark (Available non-zero means pending); until the business consumes it, the receive side does not advance — the echo service binds "consuming plaintext" and "sending plaintext back" into one loop: Send Span by Span, and on a short write (XTLS_AGAIN) pause consumption, awaiting Writable to resume — the slow client's TCP window shrinks → TLS records can't go out → plaintext consumption pauses → the receive watermark holds → memory constant. This is Chapter 62's "pump zero-blocking" promise redeemed on TLS.

Configuration sharing and lifecycle

1Assemblyidentity (Chapter 83) + server config + stream config + event table — all on the stack or shared objects

2AdmissionsStreamAccept once per TCP connection; the composite holds configuration references

3RunningN connections concurrent; callbacks execute on Engine Workers (Chapter 64)

4Closingeach connection's Close decrements the count; the listener's close sets a flag; both reaching zero releases the shared objects together

The shared configuration may not be released before the Listener and all connections close — the example orchestrates this rendezvous with atomic counters (Connections/ListenerClosed). The per-connection rule matches Chapter 82: Close/Abort → wait XTLS_STREAM_CLOSED/FAILED → Destroy.

Three TLS-specific wiring details

Three implementation joins worth knowing: handshake bytes are fed automatically by the composite (you never called Feed — after StreamAccept the handshake is "the transport's business"); the transport query xrtTlsStreamTransport(流) lends the underlying TCP stream — querying the peer address (xrtNetStreamRemote) for logging in the Open callback is the standard move; two-tier close — StreamClose goes through close_notify draining (Chapter 82), and the Close callback fires once at the composite's final state. Configuration comes in two parts: xtlsserverconfig (protocol face: identity/ALPN) and xtlsstreamconfig (stream face: buffer watermarks etc.) — protocol-transport separation of concerns down to the configuration layer.

87.3示例

First complete program: the event-driven TLS echo service

The following program comes from examples/tls/stream/main.c — a complete TLS 1.3 Echo service connectable with openssl s_client:

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> ...

What just happened. (1) The Echo main loop is the hard-backpressure template: Available → Buffer → Front 取 Span → Send(部分受理 iWritten) → Consume(iWritten) → AGAIN 即返回 — both Read and Writable enter this single no-duplication path; a slow client can never blow up memory. (2) Accept wiring: inside the TCP Accept callback, StreamAccept builds the composite in one step — this example immediately Destroys (demonstrating ownership: Accept's success hands you a reference, the event table takes over the rest; a real service lets connections run naturally to close). (3) The transport query in Open: StreamTransport lends the TCP stream to query Remote and print "TLS client open: address" — that one bridge is all there is between the TLS layer and the transport layer. (4) Lifecycle rendezvous: the Close callback decrements Connections and ListenerClose sets the flag — the main thread waits for both to reach zero before releasing the shared configuration; this is the close-side orchestration of "N connections sharing one assembly". (5) Read side by side with Chapter 67's TCP echo: the event model, Span consumption, and short-write resumption are isomorphic item by item — the only additions are the handshake and two-tier close.

Second complete program: the Stream-layer full-feature tour

The second program comes from examples/tls/stream_tour/main.c — five lines covering the Stream layer's advanced face:

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

What just happened. (1) Line 1 checks two client-admission paths: starttls (negotiated upgrade over a cleartext connection — the official answer to Chapter 87's exercise) and attach (hanging a TCP stream on directly) — plus the server's Accept, the complete set of the Stream layer's three establishment forms. (2) Line 2 demonstrates Vec batch sending (9 bytes) and Span consumption (read 4, consume 5 — consumption may run ahead of the read boundary; watermark semantics). (3) Line 3 walks asynchronous Vec writing: after the Future completes, Pending returns to zero — what Chapter 68's five send tiers' Vec/async tiers look like on a TLS stream (see stream_tour). (4) Line 4 measures the ciphertext boundary: 64 bytes plaintext → 86 bytes ciphertext (16-byte AEAD tag + record-header overhead) — a real anchor for capacity planning. (5) Line 5 runs a complete HTTP/1 request-response parse on the stream — a self-check specimen chaining Chapters 84–87's primitives, and a preview of Volume 9.

87.4契约

Establishment: server StreamAccept (one step inside the Accept callback); client attaches inline via TlsDial/DialAsync; both sides share xtlsstream's full application interface.
Events: Open/Read/Writable/Close isomorphic to the TCP stream; callbacks execute on Engine Workers; Close is the sole final-state notice (result + structured error).
Hard backpressure: stalled plaintext consumption stalls reception — Span-by-Span echo + short-write awaiting Writable is the standard echo shape; memory decoupled from the slowest client.
Transport bridge: StreamTransport lends the underlying TCP stream (read-only queries: address/state); never cache across callbacks.
Two-tier close: Close goes through close_notify draining; Abort aborts immediately; a single connection waits CLOSED/FAILED before Destroy; the service level waits "all connections + listener" reaching zero before releasing the shared configuration.
Configuration separation: xtlsserverconfig (protocol face) and xtlsstreamconfig (stream face) each independent; configuration objects share via reference counting.
Handshake automation: the Feed/Drive loop is executed by the composite — the application never sees ciphertext (stream_tour's boundary measurement is diagnostic-tool usage).
Trimming: XRT_MODULE_TLS_STREAM independent (depends on NET) — trim it away when you want only the session machine without the transport composition.

Common Mistakes

Pitfall 1: copying all the plaintext in the Read callback before processing slowly：Pitfall 1: copying all the plaintext in the Read callback before processing slowly

Symptoms: memory spikes under high throughput — every connection's Read callback copies the whole receive area into its own queue, hollowing out the backpressure mechanism.

Cause: the Stream layer's backpressure is built on "consuming plaintext (Consume)"; copying without consuming equals not consuming — the receive side keeps advancing, and your queue becomes an unbounded buffer.

错误写法

static void onRead(xtlsstream* pStream, const xnetbuf* pBuffer, ptr pData) {
	queue_push(g_MyQueue, pBuffer);   /* copied away but not Consumed: backpressure voided */
}

正确写法

static void onRead(xtlsstream* pStream, const xnetbuf* pBuffer, ptr pData) {
	/* consume as much as you can process; leave the rest for the next Read/Writable */
	while ( xrtTlsStreamAvailable(pStream) != 0 ) {
		/* ... Front takes a Span, process + Consume ... */
		if ( 下游满了 ) { return; }    
	}
}

Pitfall 2: releasing the shared configuration the moment the listener closes：Pitfall 2: releasing the shared configuration the moment the listener closes

Symptoms: the last connection's callback accesses a freed identity/config — a crash exactly when the final client disconnects.

Cause: the shared configuration's reference count sits on every connection — Listener closed does not mean connections closed; release must wait for both counts to reach zero.

错误写法

listenerClose(...) {
	xrtTlsIdentityRelease(g_Identity);  /* active connections still using it */
}

正确写法

/* atomic-counter rendezvous: release together only when Connections==0 && ListenerClosed */
if ( (xrtAtomic32Load(&Connections, XMEMORY_ACQUIRE) == 0) &&
		(xrtAtomic32Load(&ListenerClosed, XMEMORY_ACQUIRE) != 0) ) {
	release_shared_assembly();
}
/* both the Close and ListenerClose callbacks run this check */

Pitfall 3: storing the TCP stream lent by StreamTransport for long-term use：Pitfall 3: storing the TCP stream lent by StreamTransport for long-term use

Symptoms: occasional misbehavior — the cached TCP-stream pointer mismatches after the composite's internal state advances.

Cause: StreamTransport is a "transport bridge at this moment" diagnostic/query view, not a long-term handle. For transport capabilities (query address, state), take it fresh at the moment of need.

错误写法

g_SavedTcp = xrtTlsStreamTransport(pStream);   /* saved away */
/* several callbacks later */
send_on_tcp(g_SavedTcp, ...);   /* writing the transport directly, bypassing TLS: state mismatch */

正确写法

/* query fresh when needed (e.g. logging in Open); sending always goes through the TLS layer's Send */
if ( xrtNetStreamRemote(xrtTlsStreamTransport(pStream), &Remote) ) {
	log_client(Remote);
}
xrtTlsStreamSend(pStream, Data, Size, &Written);  /* plaintext through the front door */

Programming Exercises

Basic: interop with openssl

Generate a self-signed certificate and run the stream service; connect with openssl s_client -connect localhost:8443 -alpn http/1.1, type text to verify the echo; watch the server print the client address. Acceptance: the echo is byte-for-byte correct; after disconnecting (Ctrl-C or Q) the server's Close callback fires, the connection count returns to zero, and the process keeps serving the next connection.

Advanced: a plaintext proxy (observation on the decryption side)

Insert an "audit" layer before the Echo: in the Read callback tally each connection's plaintext bytes and Span count, and output a summary at Close (total bytes/average Span size/connection duration). Acceptance: 4 concurrent connections each independently tallied; data matches the openssl side's sent volume; the tally copies no plaintext (done on the zero-copy path).

Challenge: a StartTLS upgrade gateway

Implement a "cleartext handshake command first, then upgrade" gateway: after TCP Accept, don't go TLS immediately — read commands line by line (Chapter 72's line framing); on STARTTLS\r\n, call StreamAttach to upgrade to TLS (other commands echo in cleartext mode). Hint: the same connection object before and after the upgrade — compare stream_tour's starttls client path. Acceptance: cleartext-stage commands echo normally; after the upgrade, cleartext commands are refused; openssl completes the post-upgrade encrypted session via the starttls hint; connection counts and both stages' resources cleaned up correctly.

Summary

Topic	Quick reference
The composite	server `StreamAccept` (one step inside the Accept callback) / client attaches inline via `TlsDial`
Event model	Open/Read/Writable/Close isomorphic to the TCP stream; executed on Worker threads
Hard backpressure	stop consuming = stop receiving; Span echo + short-write awaiting Writable; memory decoupled from the slowest client
Application interface	Send (partial acceptance) / Available/Buffer/Front/Consume / Close / Abort / State
Transport bridge	`StreamTransport` lends the TCP stream — query fresh, never cache; sending goes through the TLS layer's front door
Configuration separation	xtlsserverconfig (protocol face) / xtlsstreamconfig (stream face); shared via reference counting
Service-level lifecycle	all connections + listener reaching zero releases the shared assembly together (atomic-counter rendezvous)
Per-connection lifecycle	Close/Abort → wait CLOSED/FAILED → Destroy (Chapter 82's rule)
Handshake automation	Feed/Drive executed by the composite; the application never sees ciphertext
Trimming	`XRT_MODULE_TLS_STREAM` independent of the session machine; stream_tour is the full-feature self-check

PreviousCh. 86 TLS（中）：服务端 NextCh. 88 会话恢复：票据、PSK 与对象契约

📖 API references in this chapter:tls (201 entries)tls_stream (67 entries)net (247 entries)

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