React/Ratchet/ZMQ中的多种订阅方法的最佳实践

Question

我尝试构建一个小实时websocket用例，用户可以登录并查看所有其他登录用户，在新用户登录或现有用户注销。

对于这种情况，当用户登录或注销时，我在UserController中使用ZMQ PUSH套接字。

UserConstroller

public function login() { 

     //... here is the auth code, model call etc... 

     $aUserData = array();// user data comes from the database with username, logintime, etc.... 

     $context = new \ZMQContext(); 
     $oSocket = $context->getSocket(\ZMQ::SOCKET_PUSH, 'USER_LOGIN_PUSHER'); // use persistent_id 
     if($oSocket instanceof \ZMQSocket) { 

      $oSocket->connect("tcp://127.0.0.1:5555"); // 
      $oSocket->send(json_encode($aUserData)); 
     } 
    } 

    public function logout() { 
     //... here is the logout code, model call etc .... 

     $aUserData = array();// user data comes from the SESSION with username, logintime, etc.... 

     $context = new \ZMQContext(); 
     $oSocket = $context->getSocket(\ZMQ::SOCKET_PUSH, 'USER_LOGOUT_PUSHER'); // use persistent_id 
     if($oSocket instanceof \ZMQSocket) { 

      $oSocket->connect("tcp://127.0.0.1:5555"); // 
      $oSocket->send(json_encode($aUserData)); 
     } 
    } 

然后我有一个推类像棘轮文档：link

在这个类有两种方法：onUserLogin和onUserLogout当然所有其他的东西，如

onSubscribe，的OnOpen，onPublish

UserInformationPusher

public function onUserLogin($aUserData) { 
     //var_dump("onUserLogin"); 
     $sUserData = json_decode($aUserData, true); 

     $oTopic = $this->subscribedTopics["user_login"]; 

     if($oTopic instanceof Topic) { 
      $oTopic->broadcast($sUserData); 
     } else { 
      return; 
     } 
    } 

    public function onUserLogout($aUserData) { 
     //var_dump("onUserLogout"); 
     $entryData = json_decode($aUserData, true); 

     $oTopic = $this->subscribedTopics["user_logout"]; 

     if($oTopic instanceof Topic) { 
      $oTopic->broadcast($entryData); 
     } else { 
      return; 
     } 
    } 

的最后一部分是WampServer/WsServer/HttpServer的与一个循环，收听传入连接。还有我的ZMQ PULL插座

RatchetServerConsole

public function start_server() { 

     $oPusher = new UserInformationPusher(); 

     $oLoop = \React\EventLoop\Factory::create(); 
     $oZMQContext = new \React\ZMQ\Context($oLoop); 
     $oPullSocket = $oZMQContext->getSocket(\ZMQ::SOCKET_PULL); 

     $oPullSocket->bind('tcp://127.0.0.1:5555'); // Binding to 127.0.0.1 means the only client that can connect is itself 
     $oPullSocket->on('message', array($oPusher, 'onUserLogin')); 
     $oPullSocket->on('message', array($oPusher, 'onUserLogout')); 


     $oMemcache = new \Memcache(); 
     $oMemcache->connect('127.0.0.1', 11211); 
     $oMemcacheHandler = new Handler\MemcacheSessionHandler($oMemcache); 

     $oSession = new SessionProvider(
      new \Ratchet\Wamp\WampServer(
       $oPusher 
      ), 
      $oMemcacheHandler 
     ); 

     //$this->Output->info("Server start initiation with memcache!..."); 
     $webSock = new \React\Socket\Server($oLoop); 
     $webSock->listen(8080, '0.0.0.0'); // Binding to 0.0.0.0 means remotes can connect 
     $oServer = new \Ratchet\Server\IoServer(
      new \Ratchet\Http\HttpServer(
       new \Ratchet\WebSocket\WsServer(
        $oSession 
       ) 
      ), 
      $webSock 
     ); 
     $this->Output->info("Server started "); 
     $oLoop->run(); 

    } 

在这个例子中，调用从登录（）或退出（）总是会调用两种方法（onUserLogin和onUserLogout）。 我无法找到一些文档，它描述了我可以在（$ event，callable $ listener）方法上使用哪些事件，有没有人有链接/知识库？ 什么是最好的方法来检查从UserController哪个方法被解雇？

• 我可以在控制器添加一些信息到$ sUserData，并在推检查这
• 我可以绑定的其他套接字连接到不同的端口（例如5554为推拉），并使用上（ ）方法
• 我可以...是否有另一个最佳实践来解决这个问题？

没有必要因为它的客户端代码工作正常

Answer 1

的密集处理与PHP的最佳实践一个月以后在WebSockets的，我从我的做法改为Crossbar.io，voryx/Thruway在PHP后台和Autobahn|JS在前端。 所有这些组件都支持WAMP V2 Websocket Standard，并且能够处理我的要求。

如果有一些要求，我可以将解决方案发布到上面的问题中，使用上述组件。

Answer 2

在你RatchetServerConsole，

删除，

$oPullSocket->on('message', array($oPusher, 'onUserLogin')); 
$oPullSocket->on('message', array($oPusher, 'onUserLogout')); 

添加，

$oPullSocket->on('message', array($oPusher, 'onUserActionBroadcast')); 

。

在你UserInformationPusher，

删除onUserLogin（）和onUserLogout（）。

添加，

public function onUserActionBroadcast($aUserData) 
{ 
    $entryData = json_decode($aUserData, true); 

    // If the lookup topic object isn't set there is no one to publish to 
    if (!array_key_exists($entryData['topic'], $this->subscribedTopics)) { 
     return; 
    } 

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

    unset($entryData['topic']); 

    // re-send the data to all the clients subscribed to that category 
    $topic->broadcast($entryData); 
} 

。

你UserConstroller（添加在$ aUserData的话题），

public function login() { 

    //... here is the auth code, model call etc... 

    $aUserData = array();// user data comes from the database with username, logintime, etc.... 

    $aUserData['topic'] = 'USER_LOGIN'; // add the topic name 

    $context = new \ZMQContext(); 
    $oSocket = $context->getSocket(\ZMQ::SOCKET_PUSH, 'my pusher'); // use persistent_id 
    if($oSocket instanceof \ZMQSocket) { 

     $oSocket->connect("tcp://127.0.0.1:5555"); // 
     $oSocket->send(json_encode($aUserData)); 
    } 
} 

public function logout() { 
    //... here is the logout code, model call etc .... 

    $aUserData = array();// user data comes from the SESSION with username, logintime, etc.... 

    $aUserData['topic'] = 'USER_LOGOUT'; // add the topic name 

    $context = new \ZMQContext(); 
    $oSocket = $context->getSocket(\ZMQ::SOCKET_PUSH, 'my pusher'); // use persistent_id 
    if($oSocket instanceof \ZMQSocket) { 

     $oSocket->connect("tcp://127.0.0.1:5555"); // 
     $oSocket->send(json_encode($aUserData)); 
    } 
} 

。

在 视图文件

最后，

<script> 
var conn = new ab.Session('ws://yourdomain.dev:9000', // Add the correct domain and port here 
    function() { 
     conn.subscribe('USER_LOGIN', function(topic, data) {     
      console.log(topic); 
      console.log(data); 
     }); 

     conn.subscribe('USER_LOGOUT', function(topic, data) {     
      console.log(topic); 
      console.log(data); 
     }); 
    }, 
    function() { 
     console.warn('WebSocket connection closed'); 
    }, 
    {'skipSubprotocolCheck': true} 
); 
</script> 

。

注意：基本思想是在推动者类中使用单个广播函数。