ChatDemo 增加 JsonProtocol协议

applications/ChatDemo/Bootstrap/BusinessWorker.php

@@ -147,9 +147,7 @@ class BusinessWorker extends Man\Core\SocketWorker
                         unset($addresses_list[$addr]);
                         Store::set($key, $addresses_list);
                         $this->notice("tcp://$addr ".$errstr." del $addr from store", false);
-                        continue;
                     }
-                    $this->notice("tcp://$addr ".$errstr, false);
                     continue;
                 }
                 unset($this->badGatewayAddress[$addr]);

applications/ChatDemo/Bootstrap/Gateway.php

@@ -79,8 +79,11 @@ class Gateway extends Man\Core\SocketWorker
      */
     public function dealInput($recv_str)
     {
-        // 这个聊天demo发送数据量都很小，一般都小于一个ip数据包，所以没有判断长度，直接返回了0，表示数据全部到达
-        // 其它应用应该根据客户端协议来判断数据是否完整
+        // 如果有Event::onGatewayMessage方法通过这个方法检查数据是否接收完整
+        if(method_exists('Event','onGatewayMessage'))
+        {
+            return call_user_func_array(array('Event', 'onGatewayMessage'), array($recv_str));
+        }
         return 0;
     }
     
@@ -565,8 +568,10 @@ class Gateway extends Man\Core\SocketWorker
      */
     protected function sendBufferToWorker($bin_data)
     {
-        $this->currentDealFd = array_rand($this->workerConnections);
-        return $this->sendToClient($bin_data);
+        if($this->currentDealFd = array_rand($this->workerConnections))
+        {
+            $this->sendToClient($bin_data);
+        }
     }
     
     /**

applications/ChatDemo/Event.php

@@ -7,9 +7,18 @@
  */
 
 require_once ROOT_DIR . '/Lib/Gateway.php';
+require_once ROOT_DIR . '/Protocols/JsonProtocol.php';
 
 class Event
 {
+    /**
+     * 网关有消息时，判断消息是否完整
+     */
+    public static function onGatewayMessage($buffer)
+    {
+        return JsonProtocol::check($buffer);
+    }
+    
    /**
     * 用户连接gateway后第一次发包会触发此方法
     * @param string $message 一般是传递的账号密码等信息
@@ -26,8 +35,8 @@ class Event
        // 不合法踢掉
        if(!$uid)
        {
-           // 返回失败
-           return GateWay::kickCurrentUser('登录失败');
+           // 踢掉
+           return GateWay::kickCurrentUser();
        }
        
        // [这步是必须的]合法，记录uid到gateway通信地址的映射
@@ -37,10 +46,10 @@ class Event
        GateWay::notifyConnectionSuccess($uid);
        
        // 向当前用户发送uid
-       GateWay::sendToCurrentUid(json_encode(array('uid'=>$uid))."\n");
+       GateWay::sendToCurrentUid(JsonProtocol::encode(array('uid'=>$uid)));
        
        // 广播所有用户，xxx connected
-       GateWay::sendToAll(json_encode(array('from_uid'=>'SYSTEM', 'message'=>"$uid come \n", 'to_uid'=>'all'))."\n");
+       GateWay::sendToAll(JsonProtocol::encode(array('from_uid'=>'SYSTEM', 'message'=>"$uid come \n", 'to_uid'=>'all')));
    }
    
    /**
@@ -55,7 +64,7 @@ class Event
        GateWay::deleteUidAddress($uid);
        
        // 广播 xxx 退出了
-       GateWay::sendToAll(json_encode(array('from_uid'=>'SYSTEM', 'message'=>"$uid logout\n", 'to_uid'=>'all'))."\n");
+       GateWay::sendToAll(JsonProtocol::encode(array('from_uid'=>'SYSTEM', 'message'=>"$uid logout\n", 'to_uid'=>'all')));
    }
    
    /**
@@ -66,7 +75,7 @@ class Event
     */
    public static function onMessage($uid, $message)
    {
-        $message_data = json_decode($message, true);
+        $message_data = JsonProtocol::decode($message);
         
         // 向所有人发送
         if($message_data['to_uid'] == 'all')

applications/ChatDemo/Protocols/JsonProtocol.php

@@ -0,0 +1,42 @@
+<?php 
+class JsonProtocol
+{
+    // 根据首部四个字节（int）判断数据是否接收完毕
+    public static function check($buffer)
+    {
+        // 读取首部四个字节
+        $buffer_data = unpack(‘Ntotal_length’, $buffer);
+        // 得到这次数据的整体长度（字节）
+        $total_length = $buffer_data['total_len'];
+        // 已经收到的长度（字节）
+        $recv_length = strlen($buffer);
+        if($total_length>$recv_length)
+        {
+            // 还有这么多字节要接收
+            return $total_length - $recv_length;
+        }
+        // 接收完毕
+        return 0;
+    }
+
+    // 打包
+    public static function encode($data)
+    {
+        // 选用json格式化数据
+        $buffer = json_encode($data);
+        // 包的整体长度为json长度加首部四个字节(首部数据包长度存储占用空间)
+        $total_length = 4 + strlen($buffer);
+        return pack('N', $total_length) . $buffer;
+    }
+
+    // 解包
+    public static function decode($buffer)
+    {
+        $buffer_data = unpack('Ntotal_length', $buffer);
+        // 得到这次数据的整体长度（字节）
+        $total_length = $buffer_data['total_len'];
+        // json的数据
+        $json_string = substr($buffer, 4);
+        return json_decode($json_string, true);
+    }
+}

applications/ChatDemo/Tests/Chat.php

@@ -1,4 +1,5 @@
 <?php
+require_once __DIR__ . '/../Protocols/JsonProtocol.php';
 ini_set('display_errors', 'on');
 error_reporting(E_ALL);
 
@@ -8,9 +9,9 @@ $sock = stream_socket_client("tcp://$ip:$port");
 if(!$sock)exit("can not create sock\n");
 
 
-fwrite($sock, 'connect');
+fwrite($sock, JsonProtocol::encode('connect'));
 $rsp_string = fgets($sock, 1024);
-$ret = json_decode($rsp_string, true);
+$ret = JsonProtocol::decode($rsp_string);
 if(isset($ret['uid']))
 {
     echo "chart room login success , your uid is [{$ret['uid']}]\n";
@@ -49,7 +50,7 @@ while(1)
                   continue;
               }
               
-              $ret = json_decode(trim($ret),true);
+              $ret = JsonProtocol::decode(trim($ret));
               
               if($ret['to_uid'] == $MYUID)
               {
@@ -69,11 +70,11 @@ while(1)
           {
              $uid = $match[1];
              $words = $match[2];
-             fwrite($sock, json_encode(array('from_uid'=>$MYUID, 'to_uid'=>$uid, 'message'=>$words))."\n");
+             fwrite($sock, JsonProtocol::encode(array('from_uid'=>$MYUID, 'to_uid'=>$uid, 'message'=>$words)));
              continue;
           }
           // 向所有用户发消息
-          fwrite($sock, json_encode(array('from_uid'=>$MYUID, 'to_uid'=>'all', 'message'=>$ret))."\n");
+          fwrite($sock, JsonProtocol::encode(array('from_uid'=>$MYUID, 'to_uid'=>'all', 'message'=>$ret)));
           continue;
 
        }