在Laravel WebSockets中实现连接生命周期管理与业务逻辑绑定

// app/WebSockets/CustomWebSocketHandler.php
<?php

namespace App\WebSockets;

use BeyondCode\LaravelWebSockets\WebSockets\PusherHandler;
use Ratchet\ConnectionInterface;
use Illuminate\Support\Facades\Log;
use SplObjectStorage; // 引入 SplObjectStorage

class CustomWebSocketHandler extends PusherHandler
{
    /**
     * @var SplObjectStorage 存储连接与业务上下文的映射
     */
    protected SplObjectStorage $connections;

    public function __construct()
    {
        parent::__construct();
        $this->connections = new SplObjectStorage();
    }

    /**
     * 当新的WebSocket连接建立时调用。
     *
     * @param ConnectionInterface $connection
     * @param \Psr\Http\Message\RequestInterface $request
     * @param string $appId
     * @return void
     */
    public function onOpen(ConnectionInterface $connection, \Psr\Http\Message\RequestInterface $request, $appId)
    {
        // 调用父类的onOpen方法，确保Pusher协议的正常初始化
        parent::onOpen($connection, $request, $appId);

        Log::info("Connection opened: {$connection->resourceId}");

        // 尝试从请求中获取业务上下文，例如用户ID或订单ID
        // 客户端可以通过WebSocket URL的查询参数传递这些信息
        $queryParams = $request->getQueryParams();
        $userId = $queryParams['user_id'] ?? null;
        $orderId = $queryParams['order_id'] ?? null;

        // 存储连接与业务上下文
        $this->connections->attach($connection, [
            'resource_id' => $connection->resourceId,
            'user_id' => $userId,
            'order_id' => $orderId,
            'connected_at' => now(),
            'channels' => [], // 用于存储该连接订阅的频道
        ]);

        if ($orderId) {
            Log::info("Order {$orderId} is now being processed by user {$userId} via connection {$connection->resourceId}");
            // 触发事件以锁定订单
            event(new \App\Events\OrderLocked($orderId, $userId, $connection->resourceId));
        }
    }

    /**
     * 当收到客户端消息时调用。
     *
     * @param ConnectionInterface $connection
     * @param \Ratchet\MessageComponent\MessageInterface $msg
     * @return void
     */
    public function onMessage(ConnectionInterface $connection, \Ratchet\MessageComponent\MessageInterface $msg)
    {
        parent::onMessage($connection, $msg);

        $payload = json_decode($msg->getPayload());

        // 进一步处理消息，例如当客户端订阅特定频道时更新上下文
        if (isset($payload->event) && $payload->event === 'pusher:subscribe' && isset($payload->data->channel)) {
            $channelName = $payload->data->channel;
            $context = $this->connections->offsetGet($connection);
            $context['channels'][] = $channelName;
            $this->connections->offsetSet($connection, $context); // 更新存储的上下文

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

            // 如果频道名包含订单ID，可以进一步提取并更新
            if (preg_match('/^private-order\.(\d+)$/', $channelName, $matches)) {
                $orderId = $matches[1];
                if ($context['order_id'] !== $orderId) {
                    Log::warning("Connection {$connection->resourceId} subscribed to order {$orderId}, but initial order was {$context['order_id']}");
                    // 可以在这里更新或处理冲突
                }
            }
        }
    }

    /**
     * 当WebSocket连接关闭时调用。
     *
     * @param ConnectionInterface $connection
     * @return void
     */
    public function onClose(ConnectionInterface $connection)
    {
        Log::info("Connection closed: {$connection->resourceId}");

        // 确保该连接存在于我们的存储中
        if ($this->connections->contains($connection)) {
            $context = $this->connections->offsetGet($connection);

            $userId = $context['user_id'];
            $orderId = $context['order_id'];

            if ($orderId) {
                Log::info("Order {$orderId} is no longer processed by user {$userId} via connection {$connection->resourceId}");
                // 触发事件以解锁订单
                event(new \App\Events\OrderUnlocked($orderId, $userId, $connection->resourceId));
            }

            // 清理连接上下文
            $this->connections->detach($connection);
        }

        // 调用父类的onClose方法
        parent::onClose($connection);
    }

    /**
     * 当连接发生错误时调用。
     *
     * @param ConnectionInterface $connection
     * @param \Exception $e
     * @return void
     */
    public function onError(ConnectionInterface $connection, \Exception $e)
    {
        Log::error("Connection error for {$connection->resourceId}: " . $e->getMessage());
        parent::onError($connection, $e);
    }
}

// app/Events/OrderLocked.php
<?php

namespace App\Events;

use Illuminate\Broadcasting\InteractsWithSockets;
use Illuminate\Foundation\Events\Dispatchable;
use Illuminate\Queue\SerializesModels;

class OrderLocked
{
    use Dispatchable, InteractsWithSockets, SerializesModels;

    public $orderId;
    public $userId;
    public $connectionId;

    public function __construct($orderId, $userId, $connectionId)
    {
        $this->orderId = $orderId;
        $this->userId = $userId;
        $this->connectionId = $connectionId;
    }
}

// app/Events/OrderUnlocked.php
<?php

namespace App\Events;

use Illuminate\Broadcasting\InteractsWithSockets;
use Illuminate\Foundation\Events\Dispatchable;
use Illuminate\Queue\SerializesModels;

class OrderUnlocked
{
    use Dispatchable, InteractsWithSockets, SerializesModels;

    public $orderId;
    public $userId;
    public $connectionId;

    public function __construct($orderId, $userId, $connectionId)
    {
        $this->orderId = $orderId;
        $this->userId = $userId;
        $this->connectionId = $connectionId;
    }
}

// config/websockets.php

return [
    // ... 其他配置

    'handler' => \App\WebSockets\CustomWebSocketHandler::class,

    // ... 其他配置
];

import Echo from 'laravel-echo';

window.Pusher = require('pusher-js');

// 假设您在后端视图中将这些ID传递给前端
const currentUserId = @json(auth()->id());
const currentOrderId = @json($order->id ?? null); // 如果在订单页面

window.Echo = new Echo({
    broadcaster: 'pusher',
    key: process.env.MIX_PUSHER_APP_KEY,
    wsHost: window.location.hostname,
    wsPort: 6001