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: 从 PHP 广播

wsCast() 是一个常规函数（资源 wsCast）。它向守护进程的内部 /message 桥发送 POST 请求，将其推送到正确的套接字上。

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

有效负载通过 phlo.js 自动传递给客户端并应用于 DOM：与您在异步 routes 中所熟知的相同 apply() 协议。

没有重试，没有死信，没有确认。 如果守护进程宕机，POST 将静默失败。为了保证交付（金融事件）：与数据库队列结合使用。

14.7: 客户端

客户端本身没有什么特别之处。在 data/app.json 中将 DOM/websocket 添加到你的资源中：

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

DOM/websocket 注入一个脚本，该脚本：

• 自动连接到 wss://<host>/websocket
• 将传入的消息直接通过 apply() 传递，inner:、outer:、class:、toast:、path: 的工作方式与 async routes 相同
• 以指数退避的方式重新连接（333 毫秒，999 毫秒，...）

如果你想从 JS 发送消息：app.websocket.send({type: 'chat.send', text: 'hi'})。

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: 已知限制

• 一次性模式每个事件需要一个 PHP 启动成本。 对于收件箱、存在和通知来说是可以的；对于高频率的遥测，运行主机作为发布版本，这样守护进程就可以从其常驻池中提供服务，从而消除该成本。一个池化的 worker 一次处理一个事件，池的大小根据需求自动调整，因此请将长时间运行的处理程序放在冷路径上（在部署后重启 workers 以便它们重新加载）。
• 负载没有版本控制， 在重构时：一次性迁移所有客户端。
• 一个进程用于整个堆栈。 守护进程拥有套接字并运行 PHP；如果它崩溃，实时（以及任何池化调度）将会在重启之前不可用。请在进程监控器下运行它。
• 没有内置加密， 使用你的反向代理进行 TLS 终止（wss://）。