Ratchet PHP WAMP-React / ZeroMQ-特定用户广播

问题描述 投票：18回答：3

注意：不是与使用this question的MessageComponentInterface相同。我改用WampServerInterface，因此此问题专门与该部分有关。我需要一个带有代码示例和解释的答案，因为我可以看到这对将来的其他人有所帮助。

尝试对单个用户进行循环推送

我正在使用Ratchet和ZeroMQ的WAMP部分，并且目前有push integration tutorial的工作版本。

我正在尝试执行以下操作：

  • zeromq服务器已启动并正在运行，已准备好记录订阅者和取消订阅者
  • 用户通过websocket协议在其浏览器中进行连接
  • 一个循环已启动，它将数据发送给请求该数据的特定用户
  • 当用户断开连接时，该用户数据的循环将停止

我有要点（1）和（2）起作用，但是我遇到的问题是第三个问题：

首先：如何将数据仅发送给每个特定用户？广播将数据发送给所有人，除非“主题”可能最终成为单个用户ID？

第二：我有一个很大的安全问题。如果我要发送哪个用户ID想要从客户端进行订阅，这似乎是我需要的，那么用户可以将变量更改为另一个用户的ID及其数据将返回。

第三：我必须运行一个单独的PHP脚本，其中包含zeromq的代码以开始实际的循环。我不确定这是否是最好的方法，我宁愿让它完全在代码库中工作，而不是单独的php文件。这是我需要排序的主要领域。

以下代码显示了我目前拥有的。

仅从控制台运行的服务器

我从字面上键入php bin/push-server.php来运行它。订阅和取消订阅将输出到此终端以进行调试。

$loop   = React\EventLoop\Factory::create();
$pusher = Pusher;

$context = new React\ZMQ\Context($loop);
$pull = $context->getSocket(ZMQ::SOCKET_PULL);
$pull->bind('tcp://127.0.0.1:5555');
$pull->on('message', array($pusher, 'onMessage'));

$webSock = new React\Socket\Server($loop);
$webSock->listen(8080, '0.0.0.0'); // Binding to 0.0.0.0 means remotes can connect
$webServer = new Ratchet\Server\IoServer(
    new Ratchet\WebSocket\WsServer(
        new Ratchet\Wamp\WampServer(
            $pusher
        )
    ),
    $webSock
);

$loop->run();

通过WebSocket发送数据的推送器

我已经省略了无用的内容，并专注于onMessage()和onSubscribe()方法。

public function onSubscribe(ConnectionInterface $conn, $topic) 
{
    $subject = $topic->getId();
    $ip = $conn->remoteAddress;

    if (!array_key_exists($subject, $this->subscribedTopics)) 
    {
        $this->subscribedTopics[$subject] = $topic;
    }

    $this->clients[] = $conn->resourceId;

    echo sprintf("New Connection: %s" . PHP_EOL, $conn->remoteAddress);
}

public function onMessage($entry) {
    $entryData = json_decode($entry, true);

    var_dump($entryData);

    if (!array_key_exists($entryData['topic'], $this->subscribedTopics)) {
        return;
    }

    $topic = $this->subscribedTopics[$entryData['topic']];

    // This sends out everything to multiple users, not what I want!!
    // I can't send() to individual connections from here I don't think :S
    $topic->broadcast($entryData);
}

开始循环使用上述Pusher代码的脚本

这是我的问题-这是一个单独的php文件，希望将来可以集成到其他代码中，但是目前我不确定如何正确使用它。我是否可以从会话中获取用户的ID？我仍然需要从客户端发送它...

// Thought sessions might work here but they don't work for subscription
session_start();
$userId = $_SESSION['userId'];

$loop   = React\EventLoop\Factory::create();

$context = new ZMQContext();
$socket = $context->getSocket(ZMQ::SOCKET_PUSH, 'my pusher');
$socket->connect("tcp://localhost:5555");

$i = 0;
$loop->addPeriodicTimer(4, function() use ($socket, $loop, $userId, &$i) {

   $entryData = array(
       'topic'     => 'subscriptionTopicHere',
       'userId'    => $userId
    );
    $i++;

    // So it doesn't go on infinitely if run from browser
    if ($i >= 3)
    {
        $loop->stop();
    }

    // Send stuff to the queue
    $socket->send(json_encode($entryData));
});

最后，要订阅的客户端js

$(document).ready(function() { 

    var conn = new ab.Session(
        'ws://localhost:8080' 
      , function() {            
            conn.subscribe('topicHere', function(topic, data) {
                console.log(topic);
                console.log(data);
            });
        }
      , function() {          
            console.warn('WebSocket connection closed');
        }
      , {                       
            'skipSubprotocolCheck': true
        }
    );
});

结论

上面的方法有效，但是我确实需要弄清楚以下几点：

  • 我如何向个人用户发送个人消息？当他们访问在JS中启动websocket连接的页面时，我是否还应该启动将内容塞入PHP（zeromq）队列中的脚本？这就是我目前正在手动执行的操作，只是感觉不正确。

  • 从JS订阅用户时，从会话中获取用户ID并从客户端发送该ID是不安全的。这可能是假的。请告诉我有一种更简单的方法，如果可以，怎么办？

php websocket zeromq ratchet reactphp
3个回答
22
投票

注意：我在这里的回答不包含对ZeroMQ的引用，因为我不再使用它了。但是，如果您需要的话，我相信您将能够弄清楚如何在此答案中使用ZeroMQ。

使用JSON

最重要的是，Websocket RFC和WAMP Spec声明要订阅的主题必须是string。我在这里有点作弊，但我仍然遵守规范：我改为通过JSON。

{
    "topic": "subject here",
    "userId": "1",
    "token": "dsah9273bui3f92h3r83f82h3"
}

JSON仍然是一个字符串，但是它使我可以通过更多数据代替“ topic”，并且PHP在另一端执行json_decode()很简单。当然，您应该验证您是否确实收到了JSON，但这取决于您的实现。

所以我在这里经过什么，为什么？

  • 主题

主题是用户订阅的主题。您可以使用它来决定将哪些数据传回给用户。

  • UserId

显然是用户的ID。您必须使用下一部分来验证此用户是否存在并被允许订阅：

  • 令牌

这应该是一种用途随机生成的令牌，在您的PHP中生成，并传递给JavaScript变量。当我说“一种用途”时，我的意思是每次您重新加载页面时（并且扩展到每个HTTP请求），您的JavaScript变量都应该在其中包含一个新令牌。该令牌应根据用户ID存储在数据库中。

然后，一旦发出Websocket请求，就将令牌和用户ID与数据库中的令牌和用户ID进行匹配，以确保用户确实是他们所说的那个人，并且他们没有弄乱JS变量。 >

注意：在事件处理程序中，您可以使用$conn->remoteAddress获取连接的IP，因此，如果有人试图进行恶意连接，则可以阻止它们（记录它们或其他内容）。

为什么这样做？

之所以有效，是因为每次建立新连接时，唯一令牌确保没有用户可以访问任何其他人的订阅数据。

服务器

这是我用于运行循环和事件处理程序的内容。我正在创建循环，完成所有装饰风格的对象的创建，并传入我的EventHandler（稍后再讲），其中也包含循环。

$loop = Factory::create();

new IoServer(
    new WsServer(
        new WampServer(
            new EventHandler($loop) // This is my class. Pass in the loop!
        )
    ),
    $webSock
);

$loop->run();

事件处理程序
class EventHandler implements WampServerInterface, MessageComponentInterface
{
    /**
     * @var \React\EventLoop\LoopInterface
     */
    private $loop;

    /**
     * @var array List of connected clients
     */
    private $clients;

    /**
     * Pass in the react event loop here
     */
    public function __construct(LoopInterface $loop)
    {
        $this->loop = $loop;
    }

    /**
     * A user connects, we store the connection by the unique resource id
     */
    public function onOpen(ConnectionInterface $conn)
    {
        $this->clients[$conn->resourceId]['conn'] = $conn;
    }

    /**
     * A user subscribes. The JSON is in $subscription->getId()
     */
    public function onSubscribe(ConnectionInterface $conn, $subscription)
    {
        // This is the JSON passed in from your JavaScript
        // Obviously you need to validate it's JSON and expected data etc...
        $data = json_decode(subscription->getId());

        // Validate the users id and token together against the db values

        // Now, let's subscribe this user only
        // 5 = the interval, in seconds
        $timer = $this->loop->addPeriodicTimer(5, function() use ($subscription) {
            $data = "whatever data you want to broadcast";
            return $subscription->broadcast(json_encode($data));
        });

        // Store the timer against that user's connection resource Id
        $this->clients[$conn->resourceId]['timer'] = $timer;
    }

    public function onClose(ConnectionInterface $conn)
    {
        // There might be a connection without a timer
        // So make sure there is one before trying to cancel it!
        if (isset($this->clients[$conn->resourceId]['timer']))
        {
            if ($this->clients[$conn->resourceId]['timer'] instanceof TimerInterface)
            {
                $this->loop->cancelTimer($this->clients[$conn->resourceId]['timer']);
            }
        }

        unset($this->clients[$conn->resourceId]);
    }

    /** Implement all the extra methods the interfaces say that you must use **/
}

基本上就是这样。这里的要点是：
  • 唯一令牌，用户ID和连接ID提供了确保一个用户看不到另一用户数据所需的唯一组合。
  • [唯一令牌意味着，如果同一用户打开另一个页面并请求订阅，他们将拥有自己的连接ID +令牌组合，因此同一用户在同一页面上的订阅将不会翻倍（基本上，每个连接都有它是自己的个人数据）。

    • 扩展名

进行任何操作之前，您应确保所有数据均已通过验证，而不是黑客尝试。使用Monolog之类的内容记录所有连接尝试，并设置发生任何严重问题时的电子邮件转发（例如，由于某人是混蛋并试图入侵您的服务器，服务器停止工作。）>

结束点

  • 验证所有内容。我不能太强调这一点。在每个请求上更改的唯一令牌是important。
  • [请记住，如果您在每个HTTP请求上重新生成令牌，并且在尝试通过websocket连接之前发出了POST请求，则必须先将重新生成的令牌传递回JavaScript，然后再尝试进行连接（否则您的令牌将无效）。
  • 记录所有内容
    • 。记录所有连接，询问主题和断开连接的人。独白对此非常有用。

要发送给特定用户，您需要ROUTER-DEALER模式而不是PUB-SUB。在《指南》的第3章中对此进行了说明。如果您使用的是ZMQ v4.0，则安全性是在线路级别上处理的，因此在应用程序中看不到它。除非您使用提供身份验证框架（zauth）的CZMQ绑定，否则仍然需要一些工作。

基本上，要进行身份验证，请在inproc：//zeromq.zap.01上安装处理程序，然后通过该套接字响应请求。适用于RFC的Google ZeroMQ ZAP； libzmq / tests / test_security_curve.cpp核心程序中也有一个测试用例。

我有类似的问题。试图弄清楚如何“侦听” rabbmitmq队列并通过ratchetphp将数据发送到websocket客户端。我不知道该怎么做。

2
投票

要发送给特定用户，您需要ROUTER-DEALER模式而不是PUB-SUB。在《指南》的第3章中对此进行了说明。如果您使用的是ZMQ v4.0，则安全性是在线路级别上处理的，因此在应用程序中看不到它。除非您使用提供身份验证框架（zauth）的CZMQ绑定，否则仍然需要一些工作。

基本上，要进行身份验证，请在inproc：//zeromq.zap.01上安装处理程序，然后通过该套接字响应请求。适用于RFC的Google ZeroMQ ZAP； libzmq / tests / test_security_curve.cpp核心程序中也有一个测试用例。

0
投票

我有类似的问题。试图弄清楚如何“侦听” rabbmitmq队列并通过ratchetphp将数据发送到websocket客户端。我不知道该怎么做。

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.