WSRouter

in package

• ZealPHP

Cross-server WebSocket routing helper.

use ZealPHP\WSRouter;
Store::defaultBackend(Store::BACKEND_REDIS);
WSRouter::init('my-app');                // server id auto-derived from hostname:pid

$app->ws('/ws', function ($server, $frame) { ... });
// In your onOpen:    WSRouter::own($clientId, $request->fd);
// In your onClose:   WSRouter::release($clientId);

// Anywhere: send to a specific client, regardless of which server holds them.
WSRouter::sendToClient($clientId, json_encode($payload));

// Or broadcast to a room channel (every subscribed server fans out locally).
WSRouter::onRoom('chat:42', function (string $payload) use ($localClients) {
    foreach ($localClients as $fd) { $server->push($fd, $payload); }
});
WSRouter::broadcast('chat:42', json_encode($payload));

Table of Contents

Constants

CLOSE_AUTH_INVALID  : mixed = 4002

Auth failed: bad token, expired session, etc.

CLOSE_AUTH_REQUIRED  : mixed = 4001

Auth required: client connected but never authenticated.

CLOSE_CAPACITY  : mixed = 4013

Server is over capacity (paired with CapacityException).

CLOSE_FORBIDDEN  : mixed = 4003

Authenticated but lacking permission for this operation.

CLOSE_GOING_AWAY  : mixed = 1001

Server is going down (or client is navigating away).

CLOSE_IDLE  : mixed = 4040

Connection idle / heartbeat missed beyond threshold.

CLOSE_INTERNAL_ERROR  : mixed = 1011

Server hit an internal error processing the request.

CLOSE_MESSAGE_TOO_BIG  : mixed = 1009

Message too big for the receiver to process.

CLOSE_NORMAL  : mixed = 1000

Normal closure (peer is closing as expected).

CLOSE_POLICY_VIOLATION  : mixed = 1008

Peer sent a message that violates server policy.

CLOSE_PROTOCOL_ERROR  : mixed = 1002

Peer sent a malformed frame / protocol violation.

CLOSE_RATE_LIMITED  : mixed = 4029

Client breached the per-client rate limit (WS-4).

CLOSE_TRY_AGAIN_LATER  : mixed = 1013

Server is temporarily overloaded — client should retry later.

CLOSE_UNSUPPORTED  : mixed = 1003

Peer sent a frame of a type the endpoint can't accept.

ROOM_CHANNEL_PREFIX  : mixed = 'ws:room:'

ROOM_TABLE  : mixed = 'ws_room_members'

SERVER_GC_INTERVAL_MS  : mixed = 60000

SERVER_HEARTBEAT_INTERVAL_MS  : mixed = 30000

SERVER_STALE_AFTER_SEC  : mixed = 90

SERVERS_TABLE  : mixed = 'ws_servers'

TABLE  : mixed = 'ws_owner'

Properties

$channelHmacSecret  : string|null

Per-channel HMAC secret (WS-3) — null disables. Set via setChannelHmacSecret.

$clientRateLimitN  : int

Per-client rate limit (WS-4) — 0 disables. Set via setClientRateLimit.

$clientRateLimitWindowSec  : int

$clientSink  : callable(string $clientId, int $fd, string $payload): void|null

Inbound-routed-message handler. Receives ($clientId, $fd, $payload) and is responsible for pushing $payload to the local fd.

$fanoutConcurrency  : int

WS-6 — max simultaneous push coroutines per room broadcast. Bounds the fan-out so a large room + message burst can't spawn an unbounded number of concurrent push coroutines (each push call is non-blocking, but the getClientInfo + dispatch path is not free, and unbounded go() per member is the keystone backpressure gap). Pushes still run concurrently (per-coroutine isolation preserved) — only the SIMULTANEITY is capped via a Coroutine\Channel token semaphore. Configure via setFanoutConcurrency.

$initialized  : bool

true once init() has wired the pub/sub subscribers and lifecycle hooks.

$localClientRooms  : array<string, array<string, true>>

Per-worker reverse index: which rooms each LOCALLY-OWNED client has joined. Lets release() (ws onClose) leave every room a client was in, so an abnormal disconnect doesn't leak ws_room_members rows / per-room SET entries until the whole server is GC'd.

$localFds  : array<string, array{fd: int, conn_id: string}>

Per-worker map of locally-owned WebSocket connections.

$localRoomMembership  : array<string, array<string, string>>

Per-worker cache of locally-owned clients by room.

$ownerCapacity  : int

Max connections per cluster — bump via initOptions() for prod.

$principalByFd  : array<int, string>

#234 — server-derived principal bound to a connection at ownAuthenticated(), keyed by fd. Lets later onMessage handlers recover the authenticated identity (principalForFd($fd)) without re-reading the session, and ties a client_id to the authenticated session rather than a client-supplied string. Reaped in release().

$rateLimitWindows  : array<string, int>

WS-7 — per-worker record of the window id each rate-limit counter is currently serving. A sliding-window limiter rotates buckets every $windowSec; the PRE-WS-7 code baked the bucket id into the counter NAME (..._{hash}_{bucket}), so on the default Atomic backend every elapsed window left a dead named Atomic forever → unbounded per-worker memory growth. WS-7 keeps the counter name STABLE (one per distinct room / client) and reuses it across windows: when this map shows the counter has rolled into a new window, the limiter reset()s it to 0 before the increment. Bounds the live Atomic count to the number of distinct rate-limited rooms + clients, not rooms × elapsed-windows.

$roomAuthorizer  : callable(string, string, string): bool|null

#234 — optional per-room authorizer. `fn(string $action, string $room, string $clientId): bool where $action ∈ {join,leave,push,read}`. When SET, every Room op consults it FAIL-CLOSED (a falsey return refuses); when null (default) the room layer is unguarded as before (BC). Wire it to your session auth — e.g. `WSRouter::roomAuthorizer(fn($a,$r,$c) => App::authChecker() && (App::authChecker())())`.

$roomMembersCapacity  : int

Max (room × member) pairs per cluster — bump via initOptions() for prod.

$roomMessageHandlers  : array<string, array<int, callable>>

User-registered message handlers per room. Each callable receives (array $msg, string $room).

$roomPresenceHandlers  : array<string, array<int, callable>>

User-registered presence (join/leave) handlers per room. Each callable receives (array $event, string $room).

$roomRateLimitN  : int

Per-room rate limit — 0 disables. Set via setRoomRateLimit($n, $windowSec).

$roomRateLimitWindowSec  : int

$serverId  : string

Cluster-unique server identifier — defaults to hostname:pid. Set by init().

$slowConsumerBytes  : int

Slow-consumer threshold. If $server->getClientInfo($fd)['send_queue_bytes'] exceeds this, the framework DROPS the message for that fd instead of letting the kernel/OpenSwoole buffer grow unbounded. Tracked in stats()['pushes_dropped_slow_consumer']. Default 4 MB; tune via initOptions(slowConsumerBytes: N).

$stats  : Stats|null

Per-worker counter struct. Surfaced via WSRouter::stats(). Lazily initialised on first access.

Methods

bootChecks()  : array<int, string>

Boot-time WS advisories — mirror of App::redisBootChecks(). Returns a list of human-readable warning strings for misconfigurations that are legal but surprising in production. App::run() (or app.php) can emit these via elog/error_log at boot. Pure + side-effect-free so it's unit-testable without booting a server.

broadcast()  : int

Publish to a room channel. Every server with a matching App::subscribe($channel, ...) handler fans out to its local clients. Returns the receiver count Redis reported.

checkClientRate()  : bool

Returns true when the client is under the rate limit (or limits disabled). Increments the (window-stable) counter atomically.

init()  : void

One-time WSRouter bootstrap. Wires the cluster-wide ownership table, room-membership table, per-server identity subscriber, and per-worker lifecycle hooks (heartbeat + GC + graceful sweep). Idempotent — subsequent calls are no-ops.

initOptions()  : void

Bump the per-cluster capacity caps BEFORE init() — these size the underlying OpenSwoole\Table segments allocated at master fork.

onlineByServer()  : array<string, int>

Per-server connection counts. Iterates ws_owner once and groups by server_id — O(N_total_clients). Useful for dashboards / debugging load-balancer imbalance; pricier than onlineCount(), so don't call on every request.

onlineCount()  : int

Total connected clients across the cluster.

onRoom()  : void

Sugar over App::subscribe($channel, ...) — use when you want the room-channel registration to read more naturally alongside the routing calls above.

own()  : string

Record that this server now owns this client's WS connection.

ownAuthenticated()  : string

Like {@see own()} but binds the connection to the authenticated session principal instead of a caller-supplied id (#234). Resolves {@see sessionPrincipal()}; throws {@see WSAuthException} when the session isn't authenticated, so an attacker can't claim another user's routing id.

principalForFd()  : string|null

The authenticated principal bound to this fd by {@see ownAuthenticated()}, or null if the connection wasn't authenticated that way. Use it to authorize onMessage-time room ops without touching the session again.

release()  : void

Clean up when a client disconnects (call from ws onClose).

room()  : Room

Construct a Room handle. The Room itself is a value object that delegates to WSRouter's static state for actual lifecycle.

roomAuthorizer()  : callable|null

Register a per-room authorizer (or read it back). `fn(string $action, string $room, string $clientId): bool, $action ∈ {join,leave,push,read}`.

sendToClient()  : bool

Send to a specific client by id, regardless of which server holds them. Returns true if the publish was issued (delivery still best-effort + subject to the conn_id verification on the receiver), false if the client isn't connected anywhere we know about.

serverId()  : string

Returns the configured server id (hostname:pid by default).

sessionPrincipal()  : string|null

Resolve the authenticated principal for the CURRENT request/connection from the session, via the same hooks the HTTP layer uses: App::authChecker() (is the session authenticated?) then App::usernameProvider() (who?). Returns null when unauthenticated, the hooks aren't wired, or no username resolves.

setChannelHmacSecret()  : void

WS-3 — set a shared HMAC secret for pub/sub channel authentication.

setClientRateLimit()  : void

WS-4 — per-client rate limit. Throttles sendToClient AND Room::push attributed to a single client. Sliding-window backed by Counter.

setFanoutConcurrency()  : void

WS-6 — bound the per-broadcast push fan-out concurrency. Default 128.

setRoomRateLimit()  : void

Configure per-room push rate limiting. Default is unlimited.

stats()  : Stats

Per-worker WSRouter counter struct. Call ->snapshot() to read the current counters as array<string,int>. Pair with Store::stats() for pool/pubsub-side metrics + App::stats() for worker/memory state.

// Inline (most concise):
WSRouter::init(
    ownerCapacity:        200_000,
    roomMembersCapacity:  1_000_000,
    slowConsumerBytes:    8 * 1024 * 1024,
);

// Or split (when you want a custom serverId / sink + tuning):
WSRouter::initOptions(ownerCapacity: 200_000, roomMembersCapacity: 1_000_000);
WSRouter::init('node-A', $myCustomSink);

WSRouter::setChannelHmacSecret(getenv('ZEALPHP_WS_HMAC') ?: null);

WSRouter::setFanoutConcurrency(256);   // up to 256 simultaneous pushes
WSRouter::setFanoutConcurrency(0);      // unbounded (legacy)