高速公路管理订阅

Question

我尝试通过可以管理多个组的 Thruway 设置一个 websocket 服务器。类似于聊天应用程序，其中每个客户端可以同时订阅一个或多个客户端并将消息广播到整个聊天室。我设法使用 Ratchet 的旧版本做到了这一点，但由于它运行不是很流畅，我想切换到 Thruway。可悲的是，我找不到任何东西来管理群组。到目前为止，作为 websocket-manager，我有以下内容，客户端正在使用当前版本的 Autobahn|js (18.x)。

有没有人知道是否可以使用类似以下内容管理订阅组？

<?php

require_once __DIR__.'/../vendor/autoload.php';

use Thruway\Peer\Router;
use Thruway\Transport\RatchetTransportProvider;

$router = new Router();
$router->addTransportProvider(new RatchetTransportProvider("0.0.0.0", 9090));

$router->start();

Answer 1

对于 ThruWay，情况与旧 Ratchet 略有不同。首先，Thruway 不是 WAMP 服务器。它只是一个路由器。所以它没有像旧 Rathcet 那样的服务器实例，它可以让您将所有服务器端功能包装起来。但它只会获取消息包，并将 route 它们发送到同一领域中的其他会话，具体取决于它们的订阅。如果您曾经使用过 socket.io，领域理念类似于不同的连接，因此您可以将会话或连接限制在单个命名空间或拆分不同套接字实例的功能，如管理、访问者等。

在autobahn 的客户端（最新版本）一旦您订阅了一个主题，然后在该主题中发布，thruway 将自动检测主题订阅者并在同一领域向他们发送消息。但是在旧棘轮中，您需要通过保留一系列可用频道来手动处理此问题，并在用户订阅时将用户添加到每个频道，并通过迭代这些用户向主题中的这些用户广播消息。这真的很痛苦。

如果您想在服务器端使用 RPC 调用而不想在客户端包含您的某些内容，您仍然可以在服务器端使用名为 internalClient 的 class。从概念上讲，内部客户端是另一个连接到您的 thruway 客户端并在内部处理某些功能而不暴露其他客户端的会话。它接收消息包并在其中执行操作，然后 returns 结果返回请求的客户端连接。我花了一段时间才明白它是如何工作的，但一旦我弄清楚背后的想法就更有意义了。

这么少的代码可以更好地解释，

在您的路由器实例中，您需要添加一个模块，（请注意，在 voxys/thruway 包示例中，内部客户端有点混乱）

server.php

require __DIR__ . "/../bootstrap.php";
require __DIR__ . '/InternalClient.php';

$port = 8080;
$output->writeln([
    sprintf('Starting Sockets Service on Port [%s]', $port),
]);
$router = new Router();

$router->registerModule(new RatchetTransportProvider("127.0.0.1", $port));   // use 0.0.0.0 if you want to expose outside world

// common realm ( realm1 )
$router->registerModule(
    new InternalClient()    // instantiate the Socket class now
);

// administration realm (administration)
// $router->registerModule(new \AdminClient());

$router->start();

这将初始化高速公路路由器并将内部客户端实例附加到它。现在在 InternalClient.php 文件中，您将能够访问实际路由以及当前连接的客户端。对于他们提供的示例，路由器不是实例的一部分，因此您只能使用新连接的会话 ID 属性。

InternalClient.php

<?php

use Thruway\Module\RouterModuleInterface;
use Thruway\Peer\Client;
use Thruway\Peer\Router;
use Thruway\Peer\RouterInterface;
use Thruway\Logging\Logger;
use React\EventLoop\LoopInterface;

class InternalClient extends Client implements RouterModuleInterface
{
    protected $_router;

    /**
     * Contructor
     */
    public function __construct()
    {
        parent::__construct("realm1");
    }

    /**
     * @param RouterInterface $router
     * @param LoopInterface $loop
     */
    public function initModule(RouterInterface $router, LoopInterface $loop)
    {
        $this->_router = $router;

        $this->setLoop($loop);

        $this->_router->addInternalClient($this);
    }

    /**
     * @param \Thruway\ClientSession $session
     * @param \Thruway\Transport\TransportInterface $transport
     */
    public function onSessionStart($session, $transport)
    {
        // TODO: now that the session has started, setup the stuff

        echo "--------------- Hello from InternalClient ------------\n";
        $session->register('com.example.getphpversion', [$this, 'getPhpVersion']);

        $session->subscribe('wamp.metaevent.session.on_join',  [$this, 'onSessionJoin']);
        $session->subscribe('wamp.metaevent.session.on_leave', [$this, 'onSessionLeave']);
    }

    /**
     * Handle on new session joined.
     * This is where session is initially created and client is connected to socket server
     *
     * @param array $args
     * @param array $kwArgs
     * @param array $options
     * @return void
     */
    public function onSessionJoin($args, $kwArgs, $options) {
        $sessionId = $args && $args[0];
        $connectedClientSession = $this->_router->getSessionBySessionId($sessionId);
        Logger::debug($this, 'Client '. $sessionId. ' connected');
    }

    /**
     * Handle on session left.
     *
     * @param array $args
     * @param array $kwArgs
     * @param array $options
     * @return void
     */
    public function onSessionLeave($args, $kwArgs, $options) {

        $sessionId = $args && $args[0];

        Logger::debug($this, 'Client '. $sessionId. ' left');

        // Below won't work because once this event is triggered, client session is already ended
        // and cleared from router. If you need to access closed session, you may need to implement
        // a cache service such as Redis to access data manually.
        //$connectedClientSession = $this->_router->getSessionBySessionId($sessionId); 
    }

    /**
     * RPC Call messages
     * These methods will run internally when it is called from another client. 
     */
    private function getPhpVersion() {

        // You can emit or broadcast another message in this case
        $this->emitMessage('com.example.commonTopic', 'phpVersion', array('msg'=> phpVersion()));

        $this->broadcastMessage('com.example.anotherTopic', 'phpVersionRequested', array('msg'=> phpVersion()));

        // and return result of your rpc call back to requester
        return [phpversion()];
    }

    /**
     * @return Router
     */
    public function getRouter()
    {
        return $this->_router;
    }


    /**
     * @param $topic
     * @param $eventName
     * @param $msg
     * @param null $exclude
     */
    protected function broadcastMessage($topic, $eventName, $msg)
    {
        $this->emitMessage($topic, $eventName, $msg, false);
    }

    /**
     * @param $topic
     * @param $eventName
     * @param $msg
     * @param null $exclude
     */
    protected function emitMessage($topic, $eventName, $msg, $exclude = true)
    {
        $this->session->publish($topic, array($eventName), array('data' => $msg), array('exclude_me' => $exclude));
    }

}

上面的示例代码中有几点需要注意， - 为了接收主题中的消息，您需要在客户端订阅该主题。 - 内部客户端可以 publish/emit/broadcast 任何主题而无需在同一领域进行任何订阅。 - broadcast/emit 函数不是原始高速公路的一部分，这是我想出来的，目的是让我的出版物更容易一些。 emit will 将消息包发送给每个订阅主题的人，发送者除外。另一方面，广播不会排除发件人。

我希望这些信息对理解这个概念有所帮助。

高速公路管理订阅

Thruway manage subscriptions

php

websocket

phpwebsocket

autobahn

thruway