Chapter 107 SSE: Server Push

Chapter Orientation

For one-way push from server to client, the Web world has two options: WebSocket (Chapters 95–97, bidirectional, a separate protocol) and SSE (Server-Sent Events — an event protocol on an HTTP response stream). xhttp's implementation philosophy in one sentence: no dedicated SSE connection, thread, queue, or network state machine — xrtHttpSseReplyCreate creates an ordinary xhttpreply with status 200, Content-Type: text/event-stream, and a one-shot body of unknown length; the actual sending is driven by the HTTP Server's one and only response state machine, TCP/TLS backpressure, and the write deadline; the returned xhttpbodystream is an independent producer end, handed to a background task or any thread to write. The client side is the same mirror: a connection layer plus an event parser (the incremental syntax machine for text/event-stream), with Last-Event-ID disconnect resume built in. The selection rule: whenever push is one-way, use SSE — the protocol is simple, it rides port 80, and automatic reconnection is built into browsers.

107.1引入

Progress push, market-data streams, log tailing, token-by-token output from AI generation — "the server keeps talking, the client only listens" covers most push needs. Bringing WebSocket to such needs is anti-aircraft guns for mosquitoes: one-way use of a bidirectional protocol, plus the full burden of handshake upgrade, heartbeats, and the close protocol. SSE's answer is "don't switch protocols": still an HTTP response, just one that never ends — on a chunked stream with Content-Type: text/event-stream, lines of event:/data:/id:/retry: fields compose events, separated by blank lines.

The engineering difficulty sits on both ends. Server: the response stream must cooperate with background producers — events come from a task thread, the connection writes on a Worker, so how does backpressure travel? Client: the event syntax is incremental (one data field may be split across multiple data: lines; events span chunk boundaries) — the parser must be streaming. xhttp gives structured answers on both ends: server-side SseSendEvent (one complete event per call, validate before encoding, never commit half an event on failure) + the xhttpbodystream producer end (AGAIN backpressure + WaitWritable recovery); the client-side parser advances field by field under cap constraints.

107.2概念

Server: riding the one and only state machine

Three design disciplines. (1) The boundary between protocol-required and application-decided: ReplyCreate sets only Content-Type; Cache-Control: no-cache, authentication, CORS, and proxy buffering policy (X-Accel-Buffering: no) are set by the application before commit with ordinary Header APIs — the SSE layer does not guess your deployment environment. (2) Atomicity: every Send fully validates and meters first, then encodes directly into one Body Stream node — no temporary event string is built, no half event committed on failure; byte and Chunk hard budgets cover concurrent reservations, queuing, and active leases. (3) Lifecycle decoupling: after Respond commits, the Reply may be destroyed while the Stream lives on — the "response object" and the "producer end" are two lifetimes.

Event structure and framing

The four elements of xhttpsseevent: Type (the event: line — the event type name), Data (the data: lines — may be multi-line), Id (the id: line — the resume cursor), Retry (the retry: line — reconnection milliseconds suggested to the client); Flags declares which elements are present. SseSendEvent frames a complete event in one call; SseSendComment (the : heartbeat comment line — keep-alive); xrtHttpBodyStreamWrite/WriteRef/WriteTake write raw bytes directly (pre-encoding events, proxy forwarding, custom extensions — the structured Writer is not a mandatory path).

Backpressure and waiting

A producer write that hits a full budget returns AGAIN — no input reference is retained, the caller retries verbatim; xrtHttpBodyStreamWaitWritable returns the shared Future representing next-generation writability — one waiter cancelling affects the other waiters, so it must not face direct cancellation (Chapter 58's shared-Future semantics). This "AGAIN + WaitWritable" is the same vocabulary as Chapter 96's WebSocket Stream.

Client: connection and parser

The client side has two layers: the connection layer (http_sse_client — initiates the GET, maintains reconnection and the Last-Event-ID header) and the parser (the incremental syntax machine for text/event-stream — field lines, event boundaries, multi-line data joining). The parser configuration caps single lines, event data, type, and ID — a structural memory bound (Chapter 99's "what to cap is decided by content structure", SSE edition: the representation body is chosen to be unbounded, the structured memory bounded). Disconnect resume: the id: sent by the server is recorded by the client and carried back as the Last-Event-ID header on reconnection — the server resumes after the cursor.

Choosing between SSE and WebSocket

107.3示例

First complete program: events, heartbeats, and direct writes — three paths

The program below is from examples/http/sse_server — the complete face of structured event sending:

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



/* 读取并打印 Reply 的完整异步正文。 */
static bool exampleReadReply(xhttpreply* pReply)
{
	xhttpbodyreader* pReader = xrtHttpBodyOpen(
		xrtHttpReplyBody(pReply)
	);
	xhttpbodychunk Chunk;
	xhttpbodystatus Status;

	if ( pReader == NULL ) {
		return false;
	}
	for ( ;; ) {
		Status = xrtHttpBodyNext(pReader, 8, &Chunk);
		if ( Status == XHTTP_BODY_DATA ) {
			(void)fwrite(Chunk.Data, 1, Chunk.Size, stdout);
			xrtHttpBodyChunkRelease(&Chunk);
			continue;
		}
		if ( Status == XHTTP_BODY_AGAIN ) {
			xfuture* pReady = xrtHttpBodyReaderWait(pReader);

			if ( (pReady == NULL) ||
				(xrtFutureWait(pReady) != XWAIT_OK) ||
				(xrtFutureState(pReady) != XFUTURE_RESOLVED) ) {
				xrtFutureDestroy(pReady);
				xrtHttpBodyReaderDestroy(pReader);
				return false;
			}
			xrtFutureDestroy(pReady);
			continue;
		}
		break;
	}
	xrtHttpBodyReaderDestroy(pReader);
	return Status == XHTTP_BODY_EOF;
}



/* 创建 SSE Reply，发送结构化事件、心跳和预封装字节。 */
int main(void)
{
	xhttpbodystreamconfig Config;
	xhttpbodystream* pStream = NULL;
	xhttpreply* pReply;
	xhttpsseevent Event;
	bool bResult;

	xrtHttpBodyStreamConfigInit(&Config);
	Config.MaxBytes = 4096;
	Config.MaxChunks = 32;
	pReply = xrtHttpSseReplyCreate(&Config, &pStream);
	if ( (pReply == NULL) || (pStream == NULL) ) {
		xrtHttpBodyStreamDestroy(pStream);
		xrtHttpReplyDestroy(pReply);
		return 1;
	}

	memset(&Event, 0, sizeof(Event));
	Event.Type = XRT_STR_LITERAL("progress");
	Event.Data = XRT_STR_LITERAL("{\"percent\":75}");
	Event.Id = XRT_STR_LITERAL("job-42:3");
	Event.Retry = 2000;
	Event.Flags = XHTTP_SSE_EVENT_TYPE |
		XHTTP_SSE_EVENT_DATA |
		XHTTP_SSE_EVENT_ID |
		XHTTP_SSE_EVENT_RETRY;
	if ( (xrtHttpSseSendEvent(
			pStream, &Event
		) != XHTTP_BODY_STREAM_OK) ||
		(xrtHttpSseSendComment(
			pStream, XRT_STR_LITERAL("heartbeat")
		) != XHTTP_BODY_STREAM_OK) ||
		(xrtHttpBodyStreamWrite(
			pStream, XRT_BYTES_LITERAL("data: ready\n\n")
		) != XHTTP_BODY_STREAM_OK) ) {
		xrtHttpBodyStreamDestroy(pStream);
		xrtHttpReplyDestroy(pReply);
		return 1;
	}

	xrtHttpBodyStreamDestroy(pStream);
	bResult = exampleReadReply(pReply);
	xrtHttpReplyDestroy(pReply);
	return bResult ? 0 : 1;
}

What just happened. (1) BodyStreamConfigInit sets MaxBytes/MaxChunks (the producer budget — the source of the AGAIN threshold), then SseReplyCreate yields the Reply and Stream object pair. (2) Three write forms fired together: SseSendEvent (a progress event with all four elements — type/data/id/retry framed at once), SseSendComment (the heartbeat comment line — keep-alive without producing an event), BodyStreamWrite (the pre-framed data: ready\n\n bytes written directly — the path for proxy/static events). (3) Read-back verification (the sample's exampleReadReply): reads the Reply's body out streaming with Chapter 108's body reader and prints chunk by chunk — the bytes the server frames are exactly the syntax the client parser eats; one sample checks both ends against each other. (4) The two-object lifetime: the Stream is Destroyed when done (EOF published), the Reply Destroyed after — consistent with the contract "after commit the Reply may be destroyed, the Stream independent" (this sample does not commit to a connection; it reads back locally instead).

Second complete program: the waiting form of client-side stream reading

The second program is from examples/http/sse_client — the standard loop of the client's consuming side:

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



/* 示例上下文只同步唯一终态，不在回调和宿主线程间共享借用视图。 */
typedef struct sse_example {
	xatomic32 Closed;
} sse_example;



/* 输出一段不保证以零字符结尾的 SSE 文本视图。 */
static void printView(cstr sName, xstrview Value)
{
	printf("%s: ", sName);
	if ( Value.Size != 0 ) {
		(void)fwrite(Value.Data, 1, Value.Size, stdout);
	}
	(void)fputc('\n', stdout);
}



/* 报告已经通过状态码和媒体类型校验的响应。 */
static bool onOpen(
	xhttpsseclient* pClient,
	const xhttpresponse* pResponse,
	ptr pData
)
{
	(void)pClient;
	(void)pData;
	printf("open: HTTP %u\n", (unsigned)xrtHttpResponseStatus(pResponse));
	return true;
}



/* 在网络 Worker 回调内消费借用消息，不把 Parser 视图带出回调。 */
static bool onMessage(
	xhttpsseclient* pClient,
	const xhttpssemessage* pMessage,
	ptr pData
)
{
	(void)pClient;
	(void)pData;
	printView("event", pMessage->Type);
	printView("id", pMessage->LastEventId);
	printView("data", pMessage->Data);
	(void)fputc('\n', stdout);
	return true;
}



/* 报告暂态断开；Client 随后按协商后的毫秒延迟自动重连。 */
static void onRetrying(
	xhttpsseclient* pClient,
	size_t iReconnect,
	uint64 iDelay,
	const xerror* pError,
	ptr pData
)
{
	(void)pClient;
	(void)pData;
	fprintf(
		stderr,
		"reconnect %zu after %llu ms: %s\n",
		iReconnect,
		(unsigned long long)iDelay,
		pError != NULL ? xrtErrorMessage(pError) : "stream ended"
	);
}



/* 发布唯一终态并唤醒示例宿主线程。 */
static void onClose(
	xhttpsseclient* pClient,
	xhttpsseclosereason Reason,
	const xerror* pError,
	ptr pData
)
{
	sse_example* pExample = (sse_example*)pData;

	(void)pClient;
	fprintf(
		stderr,
		"closed (%d): %s\n",
		(int)Reason,
		pError != NULL ? xrtErrorMessage(pError) : "normal stop"
	);
	xrtAtomic32Store(&pExample->Closed, 1, XMEMORY_RELEASE);
}



/* 连接一个 SSE URL，打印消息并等待服务端 204 或永久错误终止会话。 */
int main(int argc, char** argv)
{
	xnetengineconfig EngineConfig;
	xhttpclientconfig HttpConfig;
	xhttpsseclientevents Events;
	sse_example Example;
	xnetengine* pEngine = NULL;
	xhttpclient* pHttp = NULL;
	xhttpsseclient* pSse = NULL;
	int iExit = 1;

	if ( argc != 2 ) {
		fprintf(stderr, "usage: sse_client <http-url>\n");
		return 0;
	}
	memset(&Example, 0, sizeof(Example));
	memset(&Events, 0, sizeof(Events));
	Events.Open = onOpen;
	Events.Message = onMessage;
	Events.Retrying = onRetrying;
	Events.Close = onClose;
	Events.Data = &Example;

	xrtNetEngineConfigInit(&EngineConfig);
	pEngine = xrtNetEngineCreate(&EngineConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		goto cleanup;
	}
	xrtHttpClientConfigInit(&HttpConfig);
	pHttp = xrtHttpClientCreate(pEngine, &HttpConfig);
	if ( pHttp == NULL ) {
		goto cleanup;
	}
	pSse = xrtHttpSseConnect(
		pHttp,
		(xstrview){ argv[1], strlen(argv[1]) },
		NULL,
		&Events
	);
	if ( pSse == NULL ) {
		goto cleanup;
	}

	while ( xrtAtomic32Load(&Example.Closed, XMEMORY_ACQUIRE) == 0 ) {
		xrtSleep(10);
	}
	iExit = xrtHttpSseClientError(pSse) == NULL ? 0 : 1;

cleanup:
	xrtHttpSseClientDestroy(pSse);
	xrtHttpClientDestroy(pHttp);
	if ( pEngine != NULL ) {
		while ( !xrtNetEngineDestroy(pEngine) ) {
			xrtClearError();
			xrtThreadYield();
		}
	}
	return iExit;
}

What just happened. (1) The body-read loop's three states: XHTTP_BODY_DATA (take the chunk, consume, ChunkRelease), XHTTP_BODY_AGAIN (nothing yet — xrtHttpBodyReaderWait takes a Future and waits for arrival), EOF (the stream ends normally). The AGAIN + Wait pair is the universal skeleton of streaming consumption (symmetric to the server's AGAIN + WaitWritable). (2) The SSE client's connection layer maintains this on top: reconnect per the Retry suggestion on disconnect, the Last-Event-ID cursor carried back; the event parser turns the byte stream into structured event callbacks. (3) The companion examples/http/sse_http (the bare HTTP form without the client layer — the response stream consumed directly) covers the "want the stream, not the reconnection" scenario.

107.4契约

• No dedicated state machine: the SSE Reply is an ordinary Reply (200/text-event-stream/unknown length); sending rides the one and only response state machine with the existing backpressure and write deadline.
• Protocol/application boundary: ReplyCreate sets only Content-Type; Cache-Control/CORS/buffering policy are yours to set before commit.
• Lifecycle: after Respond the Reply may be destroyed, the Stream independent (held by a task thread/publisher); last-reference destruction = EOF after queued events; Close idempotent; Fail discards the undelivered + a stable Cause.
• Write atomicity: validate and meter first, then encode into one node; no temporary event string; no half event on failure; byte/Chunk budgets cover reservations + queuing + leases.
• Backpressure: AGAIN retains no input; WaitWritable is a shared Future (no unilateral cancellation); Write/WriteRef/WriteTake direct byte writes optional.
• Event four elements: event/data/id/retry + Flags; the comment line keeps alive; id is the resume cursor.
• Client parsing: an incremental syntax machine; single-line/data/type/ID four caps — structured memory bounded; Last-Event-ID resume.
• Consume loop: DATA (chunk + Release) / AGAIN (Wait Future) / EOF, a three-state skeleton.
• Trimming: HTTP_SSE server/client independent; the parser usable alone over custom transports.

Common Mistakes

xrtHttpBodyStreamWrite(pStream, XRT_BYTES_LITERAL(
	"data: line1\nline2\n\n"));   /* line2 without a data: prefix: a syntax error */

xhttpsseevent Event = { 0 };
Event.Data = XRT_STR_LITERAL("line1\nline2");   /* the framer splits multi-line data: */
Event.Flags = XHTTP_SSE_EVENT_DATA;
xrtHttpSseSendEvent(pStream, &Event);

while ( xrtHttpSseSend(pStream, Data) == XHTTP_BODY_STREAM_AGAIN ) {
	/* tight loop: burns CPU and may never squeeze in */
}

xhttpbodystreamresult R;
while ( (R = xrtHttpSseSend(pStream, Data)) ==
		XHTTP_BODY_STREAM_AGAIN ) {
	xfuture* pW = xrtHttpBodyStreamWaitWritable(pStream);
	if ( pW == NULL ) { break; }
	xrtFutureWaitFor(pW, Timeout);   /* wait, do not cancel the shared Future */
	xrtFutureDestroy(pW);
}
/* exiting the loop with R as OK/CLOSED: handle each */

pReply = xrtHttpSseReplyCreate(NULL, &pStream);
respond(pConn, pReply);   /* deployed behind a proxy: events batched up */

pReply = xrtHttpSseReplyCreate(NULL, &pStream);
xrtHttpReplySetHeader(pReply,
	XRT_STR_LITERAL("X-Accel-Buffering"),
	XRT_STR_LITERAL("no"));
xrtHttpReplySetHeader(pReply,
	XRT_STR_LITERAL("Cache-Control"),
	XRT_STR_LITERAL("no-cache"));
respond(pConn, pReply);

Programming Exercises

Summary