14: WebSocket

在 Phlo 中，实时功能通过 Phlo Realtime 运行，这是内置于 Phlo Daemon 的 WebSocket 服务器。一个 Node 进程拥有跨每个虚拟主机的套接字连接，并通过守护进程自己的工作池在您的 PHP 应用程序上运行每个事件。您的应用程序实现了四个钩子函数，并通过 wsCast() 从 PHP 广播。

14.1: 流媒体优先

并非所有实时更新都需要使用 WebSocket。对于单请求的一次性更新，chunk()（chunk 资源）打开一个流响应，并立即将每个调用作为一行 JSON 通过普通 HTTP 刷新；phlo.js 在命令到达时立即应用它们：

route async POST report::generate {
	foreach ($this->steps AS $i => $step){
		$step->run
		chunk(inner: arr('#progress' => $i + 1 .'/'. count($this->steps)))
	}
	apply(toast: 'Done')
}

%res->streaming = true 使得一个 route 具有与普通 apply() 调用相同的行为，而经典的 SSE (text/event-stream) 填充了相同的单向插槽。所有这些在响应流的整个过程中都保持一个 PHP worker：对于一个客户端观察的有限工作（AI token streams、导入、报告）来说，这正是合适的，而对于一个静态连接来说，它则在等待事件时显得不合适。这就是 Phlo Realtime 的用途：守护进程保持打开的套接字，您的 PHP 仅在事件到达时运行，并且广播会到达每个连接的客户端，而不仅仅是请求的那个。流媒体跟随请求；广播则跟随连接的 fleet。

14.2: Phlo Realtime 是什么

Phlo Realtime 是守护进程的 WebSocket 层（基于 ws 库构建），而不是一个单独的进程。位于端口 3001 的单个守护进程服务于整个堆栈：它接受套接字升级，通过握手的 Host 头进行路由，维护客户端注册表，运行 /message 广播桥，并将每个套接字事件分发到同一工作池中的 PHP，守护进程已经用于其他所有操作。

由于该分发是在进程内进行的，因此“套接字层”和“PHP 层”之间没有任何连接：它们是一个进程。每个事件（auth、connect、receive、close）运行匹配的 websocket::<hook> 目标，执行模式是守护进程已经为该主机执行的内容：

• 一次性（build: true 主机，即开发环境）。每个事件一个新的 PHP 进程，每次启动应用：简单明了，完全隔离，热重载。非常适合开发和低流量主机。
• 常驻池（发布主机）。守护进程的池保持工作进程处于热状态，并通过管道响应事件，无需每个事件的启动。池根据需求自动扩展和缩减，因此无需配置工作进程数量。工作安全的处理程序适用，与 FrankenPHP 工作模式相同的原则：在 static 中没有请求或用户状态，始终提交或回滚数据库工作。

无论哪种方式，每个事件都为处理程序提供完整的请求生命周期：数据库、会话、资源，一切都可以轻松访问。模式遵循主机的构建标志（在 config/daemon.js 中设置主机的条目，见下文）；处理程序代码是相同的。

14.3: 安装

守护进程是一个位于 Phlo 框架外的 Node 服务。运行它，将你的反向代理指向它，这就是整个实时设置。

git clone https://github.com/q-ainl/phlo-daemon.git <daemon>
cd <daemon>
npm install

它需要一个端口、PHP 二进制文件和主机映射：

// <daemon>/config.js
require('./phlo-daemon.js')(3001, '/usr/bin/php-zts', {
	'demo.example.nl': { app: '/var/www/demo/www/app.php', build: true },
})

第三个参数是主机映射：它将每个 Host 固定到其 app.php 路径和一个 build 标志，并在 config/daemon.js 中声明。守护进程在启动时加载它，因此它始终知道哪些主机存在，以及每个主机是一次性还是池化的。没有条目的主机会导致调度失败，从而导致升级失败。

在进程管理器下运行它（systemd / pm2 / supervisord）；phlo-daemon README 描述了 pm2 模式和 /message 桥接合同：

node <daemon>/config.js

对于生产环境，通过您的反向代理（Caddy、Nginx、FrankenPHP）将 wss:// 转发到 127.0.0.1:3001，路径为 /websocket。您的应用在 www/app.php 中声明了与 daemon 常量相同的端口：

phlo_app(
	app: __DIR__.'/../',
	daemon: 3001,
)

14.4: 应用钩子

在您的应用源代码中，您定义了四个函数；Phlo 的 websocket 资源会在它们存在时调用它们。将它们放在一个像 app.ws.phlo 的文件中：不要将文件命名为 websocket.phlo，因为该类名在加载时与引擎的 websocket 资源冲突。

function wsConnect($wsHost, $wsToken, $wsSocket){
	%log->info('ws connect', socket: $wsSocket)
	return true
}

function wsAuth($wsHost, $wsToken, $wsSocket){
	$user = %user->byToken($wsToken)
	if (!$user) error('unauthorized')
	%session->user = $user
	return true
}

function wsReceive($wsHost, $wsToken, $wsSocket, ...$data){
	$type = $data['type'] ?? null
	if ($type === 'ping') return wsCast(wsTarget: $wsSocket, pong: time())
	if ($type === 'chat.send') chat::send($data['text'], from: %session->user->id)
}

function wsClose($wsHost, $wsToken, $wsSocket){
	%log->info('ws close', socket: $wsSocket)
}

$wsSocket 是一个不透明的字符串标识符，您可以用它来精确地向这个客户端广播。

连接上下文参数按约定以 ws 为前缀（$wsHost，$wsToken，$wsSocket），与 wsCast 完全相同。这不是装饰性的：wsReceive 将 JSON 负载展开为命名参数（...$data），因此未加前缀的 $host/$token/$socket 参数将与携带 host、token 或 socket 键的负载发生致命冲突。保持前缀，您的负载键将保持自由。

14.5: 认证流程

守护进程在握手时进行身份验证，然后才接受套接字：

1. 浏览器打开 wss://<host>/websocket。该来源的 cookies，包括 token cookie，随升级请求一起发送。
2. 守护进程读取 token cookie。如果缺失，则以 401 拒绝升级。
3. 守护进程在其池上运行 websocket::auth($wsHost, $wsToken, $wsSocket)，该方法调用你的 wsAuth。
4. wsAuth 根据 %user、%session->token 或自定义查找验证 token。抛出错误 (error('unauthorized')) 以拒绝：抛出的身份验证失败将导致升级失败。成功时，套接字打开并运行 wsConnect。

token 通常来自 %user->token（每个登录用户）或 API 密钥，在页面提供时设置为 token cookie。浏览器在 WS 握手时会自动发送它；客户端不发送单独的身份验证消息。

14.6: Broadcasting from PHP

wsCast() is a regular function (resource wsCast). It does a POST to the daemon's internal /message bridge, which pushes it on to the right sockets.

wsCast(wsTarget: 'all',                  toast: 'New message received')
wsCast(wsTarget: 'socket:'.$wsSocket,    path: '/inbox')
wsCast(wsTarget: 'token:'.$token,        inner: ['#count' => $newCount])

A token is a principal, not a connection: one user with two tabs is one token with two sockets. So token:not: is how you reach other users, and wsExcept is how you reach the sender's other screens. Pass the $wsSocket your hook was handed:

function wsReceive($wsHost, $wsToken, $wsSocket, ...$data){
	wsCast(wsTarget: 'token:'.$wsToken, wsExcept: $wsSocket, inner: $data['inner'])
}

Spell the payload out there; ...$data after a named argument is a fatal PHP error.

The payload is passed through to the client and applied to the DOM automatically by phlo.js: the same apply() protocol you know from async routes.

No retry, no dead-letter, no ACK. If the daemon is down, the POST fails silently. For guaranteed delivery (financial events): combine with a DB queue.

14.7: Client side

The client itself does nothing special. Add DOM/websocket to your resources in data/app.json:

{
	"resources": [..., "DOM/websocket", "wsCast"]
}

DOM/websocket injects a script that:

• connects automatically to wss://<host>/websocket
• pipes incoming messages straight through apply(), inner:, outer:, class:, toast:, path: work the same as with async routes
• reconnects with exponential backoff (333 ms, 999 ms, ...)

If you want to send from JS: app.websocket.send({type: 'chat.send', text: 'hi'}).

To react to the connection itself, subscribe rather than assign:

const off = app.websocket.on('connect', () => app.websocket.send({want: 'state'}), element)

app.websocket.connect = fn still works, but it is a single field: a second feature assigning it replaces the first without a word. on() takes as many subscribers as you like, fires straight away when the socket is already open (so a page that mounts late still gets its moment), drops a subscriber once the element you passed leaves the document, and returns an unsubscribe function. close and error work the same way.

14.8: 迷你示例：presence

显示“谁在线”而不进行轮询。

function wsConnect($wsHost, $wsToken, $wsSocket){
	%apcu->set("presence:$wsSocket", time(), 3600)
	wsCast(wsTarget: 'all', inner: ['#online-count' => static::count()])
	return true
}

function wsClose($wsHost, $wsToken, $wsSocket){
	%apcu->delete("presence:$wsSocket")
	wsCast(wsTarget: 'all', inner: ['#online-count' => static::count()])
}

static count(){
	$keys = %apcu->keys('presence:')
	return count($keys)
}

服务器不保留任何状态；APCu 计算每个主机的套接字数量。在 PHP 重启时，缓存会自动清空，这很好，因为空的存在是一个可接受的降级状态。

14.9: Known limitations

• One-shot mode costs a PHP startup per event. Fine for inbox, presence and notifications; for high-frequency telemetry run the host as a release build so the daemon serves it from its resident pool, which removes that cost. A pooled worker handles one event at a time and the pool sizes itself to demand, so keep long-running handlers off the hot path (restart the workers after a deploy so they reload).
• No versioning on payloads, when refactoring: migrate all clients at once.
• One process for the stack. The daemon owns the sockets and runs the PHP; if it crashes, realtime (and any pooled dispatch) is down until restart. Run it under a process supervisor.
• No built-in encryption, use your reverse proxy for TLS termination (wss://).

Reference. The realtime and DOM resources are documented per node in the Manual, generated from the resource files so it never drifts from the code.

最近更新于 2026年8月23日