定制Laravel Websockets连接生命周期与状态管理实践

// app/Websockets/CustomWebSocketHandler.php

namespace App\Websockets;

use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
use BeyondCode\LaravelWebSockets\WebSockets\WebSocketHandler;
use Ratchet\ConnectionInterface;
use SplObjectStorage; // 用于存储连接相关数据，适用于单服务器部署
use Illuminate\Support\Facades\Log;
// use Illuminate\Support\Facades\Redis; // 对于多服务器部署，推荐使用Redis

class CustomWebSocketHandler extends WebSocketHandler
{
    /**
     * 存储连接到其订阅的频道列表的映射。
     * @var SplObjectStorage<ConnectionInterface, array<string>>
     */
    protected SplObjectStorage $connectionChannels;

    /**
     * 存储连接到其应用层上下文的映射 (例如: user_id, 关联的资源ID等)。
     * @var SplObjectStorage<ConnectionInterface, array<string, mixed>>
     */
    protected SplObjectStorage $connectionContext;

    public function __construct(ChannelManager $channelManager)
    {
        parent::__construct($channelManager);
        $this->connectionChannels = new SplObjectStorage();
        $this->connectionContext = new SplObjectStorage();
    }

    /**
     * 当新的WebSocket连接建立时触发。
     */
    public function onOpen(ConnectionInterface $connection)
    {
        parent::onOpen($connection);
        Log::info("Connection opened: {$connection->resourceId}");

        // 初始化该连接的频道和上下文存储
        $this->connectionChannels->attach($connection, []);
        $this->connectionContext->attach($connection, []);

        // 在此阶段，通常还没有明确的应用层上下文（如用户ID或订单ID）。
        // 这些信息通常在订阅私有频道后才能获取。
    }

    /**
     * 当连接接收到客户端发送的消息时触发。
     * 我们在此拦截 'pusher:subscribe' 消息以获取频道信息。
     */
    public function onMessage(ConnectionInterface $connection, $msg)
    {
        $message = json_decode($msg, true);

        // 检查是否是订阅频道的消息
        if (isset($message['event']) && $message['event'] === 'pusher:subscribe') {
            $channelName = $message['data']['channel'] ?? null;

            if ($channelName) {
                // 将频道添加到该连接的订阅列表中
                if (!$this->connectionChannels->contains($connection)) {
                    $this->connectionChannels->attach($connection, []);
                }
                $channels = $this->connectionChannels[$connection];
                if (!in_array($channelName, $channels)) {
                    $channels[] = $channelName;
                    $this->connectionChannels[$connection] = $channels;

                    Log::info("Connection {$connection->resourceId} subscribed to channel: {$channelName}");

                    // 针对私有频道，Laravel Websockets会在认证成功后将用户对象存储在 ConnectionInterface->app->user 中
                    $user = $connection->app->user ?? null;
                    if ($user) {
                        $context = $this->connectionContext[$connection];
                        $context['user_id'] = $user->id;
                        $this->connectionContext[$connection] = $context;
                        Log::info("Connection {$connection->resourceId} associated with user ID: {$user->id}");
                    }

                    // 如果是特定资源频道 (例如 'private-order.{order_id}')
                    if (str_starts_with($channelName, 'private-order.')) {
                        $orderId = (int) substr($channelName, strlen('private-order.'));
                        $this->lockOrder($orderId, $connection->resourceId, $user->id ?? null);

                        // 也可以将订单ID存储到连接上下文中
                        $context = $this->connectionContext[$connection];
                        if (!isset($context['order_ids'])) {
                            $context['order_ids'] = [];
                        }
                        $context['order_ids'][] = $orderId;
                        $this->connectionContext[$connection] = $context;
                    }
                }
            }
        }

        // 务必调用父类的 onMessage 方法，以确保标准的WebSocket操作（如实际的频道订阅处理）得到执行
        parent::onMessage($connection, $msg);
    }

    /**
     * 当WebSocket连接关闭时触发。
     * 我们在此执行资源解锁和清理工作。
     */
    public function onClose(ConnectionInterface $connection)
    {
        Log::info("Connection closed: {$connection->resourceId}");

        // 获取该连接订阅的所有频道，并执行解锁逻辑
        if ($this->connectionChannels->contains($connection)) {
            $channels = $this->connectionChannels[$connection];
            foreach ($channels as $channelName) {
                // 如果是特定资源频道，则执行解锁操作
                if (str_starts_with($channelName, 'private-order.')) {
                    $orderId = (int) substr($channelName, strlen('private-order.'));
                    $this->unlockOrder($orderId, $connection->resourceId);
                }