Chapter 144 Project 5: A WebSocket Real-Time Push Service

Chapter Orientation

Volume 13's Project 5: a real-time push service (pushd) — a WebSocket long-connection gateway: after connecting, clients subscribe to channels ({"op":"sub","ch":"metrics"}), and the server pushes backend-produced messages to every channel member (broadcast); it supports kicks (kick), channel-member queries, and graceful shutdown (the full Seal-then-clear flow). It applies the four xws chapters (109–112) together: the routing endpoint (Chapter 113's xrtWsServerRoute — the secure entrance with Origin policy and the authorize callback), connection groups (Chapter 112's xwsgroup — membership / snapshot broadcast / Seal half-close), the send path (Chapter 111's writer/references/compression — broadcast's multicast primitives), the connection lifecycle (Chapter 110's pause/resume/close protocol). The new system constraint is long-connection concurrency: the lifecycle management of hundreds of simultaneously live connections, the fan-out amplification of broadcast, and the impact of slow clients on the whole — Projects 1/2/4's lessons plus this one converge into the full stack in the static server (Chapter 145).

144.1引入

The requirement scene: a monitoring dashboard must show service metrics live — polling (one pull per second) is laggy and wasteful; SSE (Chapter 107) suffices for one-way, but the client needs subscription semantics across channels and bidirectional control (sub/unsub/kick) — WebSocket's bidirectional long connection is the natural form. pushd's interface design: connection protocol (upgrade at /ws, subprotocol push.v1); subscription messages (JSON: sub/unsub with channel names); push messages (JSON: channel + payload + sequence — clients check the sequence for losses); management actions (kick a connection, list channel members — the operations interface). Acceptance criteria: (1) 100 concurrent connections, three channels, broadcasts fully delivered (no misses — sequence reconciliation); (2) a slow client (receiving but not reading) does not drag fast clients (isolation verified); (3) kick takes effect immediately (later messages never reach the kicked); (4) graceful shutdown: after Seal new connections refused, in-flight messages delivered, exit only after all connections close (final-state count reconciliation); (5) broadcast memory decoupled from connection-count × message-size (reference send — verified with stats); (6) the service stays stable under backpressure (broadcast never blocks on a single full connection).

144.2概念

Architecture: three-layer assembly

The subscription-set placement decision: the connection group manages only "the set of connections" (Chapter 112's boundary — no message encoding, no business); the channel→connection-set mapping is business state — an independent multi-group structure (one xwsgroup per channel) or Map<channel, group>. Decision: one group per channel — the group's concurrency semantics (unique membership/references/snapshots) are exactly what channel membership needs, obtained free; the alternative Map<channel, connection array> would hand-roll all the concurrency semantics — rejected.

Module selection and rationale

The broadcast pipeline: from a backend message to N clients

The sequence number's role: monotonically increasing per channel — clients detect losses (receiving 3 then 5 — 4 was lost: an active re-sync). Reference send's multicast upgrade (Chapter 111): one payload, each of the N connections' queues holding a reference — the release callback counts down (the true release only after the last write); GroupTextAsync is exactly this shape inside. The three slow-connection grades: skip + mark slow (default — a lost frame is acceptable, the client re-syncs); kick (a configured threshold — cleaning chronically degraded connections); unbounded waiting (not provided — a single connection dragging everyone violates acceptance 2/6).

The subscription protocol and message handling

A message arrives (Chapter 110's event) → JSON parse (Chapter 32) → dispatch by the op field: sub (validate the channel name → GroupAdd into the group — create on first); unsub (GroupRemove + reference return); heartbeat (client ping — the connection layer's automatic Pong already handles it (Chapter 110); the application layer may optionally echo online status). Illegal messages: a parse failure returns an error frame (a JSON error code) without disconnecting (protocol leniency — one corrupt frame does not mean a bad connection); an illegal channel name returns an error — every error locatable (Chapter 4's error model, application-message edition).

The graceful-shutdown flow

Chapter 112's shutdown template fully unfolded in pushd: step one stop accepting (listener-layer Drain — new HTTP upgrades no longer admitted); step two Seal every channel group (GroupSeal — idempotent; new subscriptions blocked, existing members send and receive as usual); step three the farewell broadcast (a "server shutting down" frame — a clear signal to clients, never a silent drop); step four close connection by connection (ConnClose with NORMAL code — the protocol-driven goodbye); step five wait for the Close-callback count to reach zero (all connections final); step six destroy groups and routes (Release's Data lifecycle concludes here — Chapter 113's "exactly once after the Router and all connections exit"); step seven stop the Engine. Acceptance 4's reconciliation points: closing frames delivered (clients received shutting down) + zero leaks (the groups' member counts at zero) + exit code 0.

Against the earlier projects' lessons

The twelve lessons (inventoried in Chapter 143) replay in long-connection form plus additions: acceptance first (six items — concurrency and timing entries rising); shape decides the container (channel = group, subscription set = Map, message = JSON — three shapes, three containers); views are use-and-discard (message-event views parsed and dispatched on the spot — no buffering); layered swappability (swapping the transport for TLS or the serialization for XLON is single-layer); platform facts enter design (Origin policy and subprotocol negotiation are WS's "platform facts"); budget first (broadcast fan-out = message rate × connection count — the budget basis for backpressure and slow policy); four-quadrant testing (concurrent tests become the main battlefield: 100 connections × three channels × sequence reconciliation). Two additions: fan-out is an amplifier (1 backend message × 1000 connections = 1000 sends — every nanosecond of the broadcast path is amplified; reference send is not an optimization but a necessity); long-connection state cleanup is the final test (a short connection's cleanup happens naturally after each request; a long connection's cleanup happens only at shutdown and kick — leaks are stealthier, and final-state count reconciliation is the only reliable acceptance).

144.3示例

First complete program: channel-group membership management

The program below is from examples/websocket/group — the integration template of channel membership:

#include <stdio.h>

#include <xws.h>



/* Upgrade 成功后把借用连接加入应用连接组。 */
static void connectionOpen(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupAdd(pGroup, pConnection);
}



/* 唯一 Close 回调中移除连接，归还连接组持有的引用。 */
static void connectionClose(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupRemove(pGroup, pConnection);
}



/* 示例保留回调形状，实际服务器可直接把组放入路由上下文。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);

	if ( pGroup == NULL ) {
		return 1;
	}
	(void)connectionOpen;
	(void)connectionClose;
	printf("WebSocket connection group is ready\n");
	xrtWsGroupDestroy(pGroup);
	return 0;
}

$ 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/group/main.c -lws2_32 -lbcrypt -ladvapi32 -liphlpapi
WebSocket connection group is ready

What just happened. (1) The connectionOpen/Close pair is pushd's subscribe/unsubscribe kernel: on upgrade success GroupAdd (join the channel), in the sole Close callback GroupRemove (return the reference) — a connection's group lifecycle synchronizes strictly with the connection lifecycle (no unsubscribe leak possible — Close is the only final state, guaranteed exactly once). (2) pushd adds channel routing (multiple groups) and the message protocol on top — membership's concurrency semantics (idempotent repeated Add / capacity refusal / stable snapshots) all inherited from the group. (3) A real server puts the group into the route's Data (Chapter 113) — this sample locks in "the shape of integration".

Second complete program: the async broadcast operation object

The second program is from examples/websocket/group_future — the waitable broadcast:

#include <stdio.h>

#include <xws.h>



/* 展示空组广播也会返回可等待、可检查的完整操作对象。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);
	xwsgroupop* pOperation;

	if ( pGroup == NULL ) {
		return 1;
	}
	pOperation = xrtWsGroupTextAsync(
		pGroup,
		XRT_STR_LITERAL("service-ready")
	);
	if ( (pOperation == NULL) ||
		(xrtWsGroupOpWait(pOperation) != XWAIT_OK) ) {
		xrtWsGroupOpDestroy(pOperation);
		xrtWsGroupDestroy(pGroup);
		return 1;
	}
	printf(
		"members=%zu accepted=%zu rejected=%zu\n",
		xrtWsGroupOpCount(pOperation),
		xrtWsGroupOpAccepted(pOperation),
		xrtWsGroupOpRejected(pOperation)
	);
	xrtWsGroupOpDestroy(pOperation);
	xrtWsGroupDestroy(pGroup);
	return 0;
}

$ 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/group_future/main.c -lws2_32 -lbcrypt -ladvapi32 -liphlpapi
（空组广播操作完成自检通过后正常退出）

What just happened. (1) xrtWsGroupTextAsync returns an operation object — the three steps initiate/wait/release: pushd's broadcast call shape (the producer fires and doesn't wait — the timer rhythm drives; the management interface waits — confirm delivery after a kick). (2) An empty group is also a complete operation (zero connections = zero successes — aggregate semantics self-consistent). (3) The slow-connection policy point sits here: the operation object does not wait for any single slow connection (complete once all accepted — a slow one's AGAIN is handled by its own connection's backpressure) — "per-connection independent acceptance" is precisely the mechanism behind acceptance 2/6.

Implementation walkthrough: the broadcast and subscription core source

pushd's two core loops — subscription handling and channel broadcast — in source form (key segments walked over the teaching samples):

/* message event: JSON protocol dispatch (sub/unsub) */
static void on_message(xwsconn* Conn, const uint8_t* Data,
	size_t Size, ptr Ctx)
{
	pushd* S = Ctx;
	xvalue Msg;
	xstrview Op, Ch;

	/* parse failure: reply with an error frame, keep the connection (the lenient contract) */
	xvalue* V;
	if ( !xrtJsonParse(Data, Size, &Msg) ||
		((V = xrtValueObjectGet(Msg,
			XRT_STR_LITERAL("op"))) == NULL) ||
		!xrtValueGetString(V, &Op) ) {
		send_error(Conn, "bad message");
		return;
	}
	if ( ((V = xrtValueObjectGet(Msg,
			XRT_STR_LITERAL("ch"))) == NULL) ||
		!xrtValueGetString(V, &Ch) ||
		!channel_name_valid(Ch) ) {
		send_error(Conn, "bad channel");
		return;
	}

	if ( xrtStrEqual(Op, XRT_STR_LITERAL("sub")) ) {
		xwsgroup* Gr = pushd_channel(S, Ch);  /* fetch or create */
		xrtWsGroupAdd(Gr, Conn);               /* idempotent */
		send_ok(Conn, "subscribed");
	} else if ( xrtStrEqual(Op, XRT_STR_LITERAL("unsub")) ) {
		xwsgroup* Gr = pushd_find(S, Ch);
		if ( Gr != NULL ) { xrtWsGroupRemove(Gr, Conn); }
		send_ok(Conn, "unsubscribed");
	}
}

/* channel broadcast: snapshot + reference send + slow-connection skip */
static void pushd_broadcast(pushd* S, xstrview Ch, str Payload)
{
	xwsgroupsnapshot* Snap;
	size_t i;

	xwsgroup* Gr = pushd_find(S, Ch);
	if ( (Gr == NULL) || xrtWsGroupSealed(Gr) ) { return; }
	Snap = xrtWsGroupSnapshotCreate(Gr);   /* stable snapshot */
	for ( i = 0; i < xrtWsGroupSnapshotCount(Snap); i++ ) {
		xwsconn* C = xrtWsGroupSnapshotGet(Snap, i);
		xwsgroupop* Op = xrtWsGroupTextAsync(Gr, view(Payload));
		if ( Op == NULL ) { S->SlowCount++; }  /* skip + mark slow */
		else { xrtWsGroupOpDestroy(Op); }
	}
	xrtWsGroupSnapshotDestroy(Snap);       /* return the member references */
}

Walkthrough points. (1) on_message's dispatch order: parse the whole frame first (a bad frame gets an error, no disconnect), then field validation, then action — each level's error independently locatable (bad message / bad channel / success receipt). (2) pushd_channel fetches or creates the group — the first subscription creates it, GroupAdd is idempotent (repeated subs to one channel safe — Chapter 112's unique-membership semantics cashed). (3) pushd_broadcast's snapshot loop is the positive form of pitfalls 1/2: snapshot-stable iteration, group-level async operations, failure-skip counting — three contract pieces in place at once. (4) The shutdown path's GroupSealed check: after Seal, broadcasts fall silent naturally (the farewell broadcast goes out before the Seal — order is semantics).

The acceptance matrix and concurrent-test design

The six criteria's test matrix (Chapter 133's four quadrants, long-connection edition): positive — 100 connections, three channels × 1000-frame broadcast with sequence reconciliation (acceptance 1); concurrent races — broadcast concurrent with subscribe/unsubscribe (snapshot semantics under a race condition, measured), broadcast concurrent with kick (no reach after Remove — acceptance 3); slow client — 1 slow (receiving, not reading) + 99 fast (acceptance 2: the fast ones' sequences unaffected); backpressure — inject production rate > consumption rate (acceptance 6: service stable, the slow count rising); shutdown — shutdown with 50 live connections (acceptance 4: shutting-down frames all delivered, counts at zero, exit code 0); resources — OOM point by point (an allocation failure on the broadcast path — skip that frame, don't crash). The tests' client side also needs a concurrency framework — Chapter 59's executor drives 100 virtual clients (connect/subscribe/receive/reconcile) — the test code itself is asynchronous and concurrent, and its correctness rests on sequence reconciliation (never on timing assumptions).

Mapping onto the book's knowledge map

pushd's bricks: Chapter 4 (error model — message-level error frames), Chapter 18 (Map — the channel-name-to-group routing), Chapter 25 (strings — channel-name validation and op comparison), Chapter 32 (JSON — the subscription protocol's codecs), Chapters 55–59 (concurrency/executor — the test-client framework), Chapter 64 (event ports — what the Engine drives underneath), Chapter 67 (TCP — the connection's transport base), Chapter 94 (HTTP upgrade — WS's handshake layer), Chapters 95–97 (the WebSocket protocol core — frames/streams/close), Chapters 104–105 (the HTTP server — the Router and middleware entrance), Chapters 110–113 (the four xws chapters — this project's direct toolbox). Sixteen chapters each in place inside pushd — Project 5 is currently the widest-coverage project (the static-server finale will be wider). The knowledge map's verification meaning: no chapter's knowledge is isolated — each has been combined in some real shape.

144.4契约

• Endpoint security: the xrtWsServerRoute fixed endpoint — Origin three-tier default (SAME_HOST_OR_ABSENT) + optional Authorize token check; Open borrows the connection, long-term holding requires Ref before storing (Chapter 113's contract).
• Channel as group: one xwsgroup per channel — membership semantics (idempotent Add / capacity refusal / stable snapshot) from the group; subscription set and membership management separate (business to the application layer).
• Broadcast pipeline: snapshot (stable iteration + references) → per-member reference send → snapshot destroy returning references; one payload shared by N connections (memory decoupled from fan-out).
• Slow clients, three grades: skip + mark slow (default) / kick (threshold) / no unbounded waiting — a single connection never drags the whole.
• Sequence protocol: monotone per channel — clients check the sequence and actively re-sync (the operations dividend of restart-without-loss semantics).
• Message leniency: a single-frame parse failure returns an error frame without disconnect; every error locatable.
• Shutdown in seven steps: stop accepting → Seal (idempotent) → farewell broadcast → close per connection → counts to zero → destroy groups and routes → stop the Engine — final-state reconciliation.
• Data lifecycle: the route Release runs exactly once after the Router and all connections exit (Chapter 113's contract).
• Kick semantics: GroupRemove first (leave the group), then ConnClose — after leaving, broadcasts never reach the kicked (the mechanism behind acceptance 3).

Delivery form and the operations interface

pushd's delivery and operations design: deployment form — a standalone process (single-header or static-library link — Chapter 132's choice); listen address / channel capacity / slow threshold read from the command line or the environment (Chapter 43) — the tool's configuration face stays minimal (three parameters suffice for deployment; complex configuration is configd's domain). Operations interface — a management endpoint (HTTP: list channels / member counts / slow counts / kicks) hung on another path of the same Router (Chapter 105's HTTP routes coexisting with the WS endpoint — Chapter 113's "one Router, two endpoint kinds"); output JSON (Chapter 32) for scripts to consume. Observability — connection count / channel count / broadcast rate / slow count periodically emitted (Chapter 38's logger or direct stdout JSON Lines — Chapter 139's file_json shape); stats (Chapter 6) watches the memory watermark. Graceful restart — the seven shutdown steps + systemd-style supervision: across the restart window clients reconnect (Chapter 110: reconnecting is the client's duty) + the sequence re-sync backfills lost frames — restart without losing semantics is the sequence protocol's operations dividend.

Re-examining the boundary against SSE push

With pushd done, Chapter 107's selection judgment deserves re-examination: when is SSE enough, when is WebSocket required? Of pushd's three requirements, channel subscription (the client telling the server what it wants) and kick management (bidirectional control) are beyond SSE — SSE can only push server-side; the client has no uplink (subscription would ride URL parameters — unrealistic with many channels); sequenced push itself SSE can do too (SSE also has the id field and Last-Event-ID resume — Chapter 107's mechanism). So the dividing line sharpens: pure downlink push uses SSE (simpler — port 80, automatic browser reconnect); client uplink control (subscription/acknowledgment/custom heartbeats) needs WebSocket. pushd's subscription semantics put it on the WS side — but if the requirement degrades to "broadcast the same stream to everyone" (a site-wide notice), SSE is the lighter answer. Selection review (looking back at the alternative after implementing one solution) is part of design ability — it makes the next selection faster and sharper.

Stress methodology and capacity planning

Chapter 136's method landed for a long-connection service: the stress question ("at 1000 connections × 3 channels × 100 frames/second of broadcast, what are the P99 frame latency and memory watermark?" — write the question before building the experiment); the load model — virtual clients (executor-driven: connect / subscribe / receive / reconcile, four phases) + producers (timers at fixed rates) — both ends deterministic scripts (no real-network dependency, local loopback — Chapter 104's mock idea, stress edition); measurement — frame-latency distribution (a send-to-receive delay histogram — Chapter 42 timestamps), memory watermark (periodic stats sampling), slow-count trajectory; sweeps — a workload of connection-count gradients (100/500/1000) × a rate gradient (10/100/1000 frames per second) — the inflection point (which dimension saturates first: memory? fan-out CPU? Engine Workers?) names the scaling direction. Capacity planning output — a single instance's recommended load band (say "≤800 connections and ≤200 frames/second with P99 < 50ms") + actions beyond it (add instances / lower the frame rate / fragment channels).

Retrospective: Project 5's new lessons

pushd's additions atop the twelve: design fan-out before optimizing (the broadcast path's reference send/snapshot/skip are design decisions, not performance patches — settle them at architecture time); long-connection final-state reconciliation (the cleanup verification of three state classes — connections/groups/subscription sets — can only rest on counts reaching zero — Chapters 110–112's per-layer "exactly once" converge into one reconciliation table at project level); the sequence number is a cheap self-healing protocol (one monotone integer buys loss detection + re-sync — two orders of magnitude cheaper than an acknowledgement-retransmission mechanism; good enough is good protocol); the test client is itself a system under test (a concurrent test framework's correctness cannot rest on timing assumptions — sequence reconciliation applies to the test code itself); the operations interface is part of the product (listing channels/kicks/slow counts are not "add later" but part of the delivery definition — Chapter 104's five timeouts likewise: operability is decided at design time). Five more, seventeen cumulative — the final project (the static-server finale) will test their resilience under full-stack combination.

Common Mistakes

n = xrtWsGroupCount(pGroup);
for ( i = 0; i < n; i++ ) { 发(第 i 个成员); }   /* joins/leaves during the loop: scrambling */

xwsgroupsnapshot* Snap = xrtWsGroupSnapshotCreate(pGroup);
for ( i = 0; i < xrtWsGroupSnapshotCount(Snap); i++ ) {
	发送(xrtWsGroupSnapshotGet(Snap, i));
}
xrtWsGroupSnapshotDestroy(Snap);

for ( each member ) {
	while ( send(conn, msg) == AGAIN ) { wait(); }  /* the slow one drags everyone */
}

/* the group operation object: returns once all accepted; a slow one's AGAIN belongs to its own connection backpressure */
xwsgroupop* Op = xrtWsGroupTextAsync(pGroup, msg);
xrtWsGroupOpDestroy(Op);   /* fire and go (or wait on the aggregate as needed) */

stop_listener();
xrtWsGroupDestroy(pGroup);   /* in-flight upgrades / connections unclosed: dangling and deadlock */

stop_listener();
xrtWsGroupSeal(pGroup);          /* 1. block new */
broadcast("shutting down");       /* 2. signal */
close_all_members(pGroup);       /* 3. protocol-driven close */
wait_close_count_zero();         /* 4. final-state count */
xrtWsGroupDestroy(pGroup);       /* 5. destroy */
xrtNetEngineDestroy(pEngine);    /* 6. stop the Engine last */

Programming Exercises

Summary