debug talk

centra_srv.php

@@ -6,7 +6,6 @@ require_once (BASE_ROOT_PATH . '/fooder.php');
 require_once (BASE_ROOT_PATH . '/helper/search/srv_base.php');
 require_once (BASE_ROOT_PATH . '/helper/search/server.php');
 require_once (BASE_ROOT_PATH . '/helper/search/processor.php');
-require_once (BASE_ROOT_PATH . '/helper/search/event_handler.php');
 require_once (BASE_ROOT_PATH . '/helper/search/util.php');
 require_once (BASE_ROOT_PATH . '/helper/category_helper.php');
 require_once (BASE_ROOT_PATH . '/helper/brand_helper.php');

helper/algorithm.php

@@ -248,6 +248,17 @@ class uniquer
         }
         return true;
     }
+    public function remove_value($val)
+    {
+        if(algorithm::binary_search($this->mContainer,$val,$this->mCompare) != false) {
+            $pos = algorithm::lower_bonud($this->mContainer,$val);
+            algorithm::array_erase($this->mContainer,$pos);
+            return true;
+        } else {
+            return false;
+        }
+    }
+
     public function values() {
         return $this->mContainer;
     }

helper/room/base_room.php

@@ -127,11 +127,15 @@ abstract class base_room extends base_info
 
         if($type == false || $content == false) return false;
 
-        $this->record_message($userinfo['userid'],$type,$content);
-        $this->relay_broadcast('message',['from' => $userinfo,'type' => $input['type'],'content' => $content]);
-        $this->relay_users([$user],'message',['from' => $userinfo,'type' => $input['type'],'content' => $content]);
-
-        return true;
+        $msgid = $this->record_message($userinfo['userid'],$type,$content);
+        if($msgid > 0) {
+            $this->relay_broadcast('message',['msgid' => $msgid, 'from' => $userinfo,'type' => $input['type'],'content' => $content]);
+//            $this->relay_users([$user],'message',['msgid' => $msgid,'from' => $userinfo,'type' => $input['type'],'content' => $content]);
+            return [''];
+        }
+        else {
+            return false;
+        }
     }
     //////////////////////////////////////消息验证、返回、广播//////////////////////////////////////////////////////////////
     protected function validate_type($type)
@@ -158,7 +162,7 @@ abstract class base_room extends base_info
     protected function record_message($userid,$type,$content)
     {
         $mod_room = Model('room');
-        $mod_room->addRoomMsg(['room_id' => $this->mRoomid,'member_id' => $userid, 'type' => $type,'msg' => $content,'add_time' => time()]);
+        return $mod_room->addRoomMsg(['room_id' => $this->mRoomid,'member_id' => $userid, 'type' => $type,'msg' => $content,'add_time' => time()]);
     }
 
     public function relay_users_msgs()
@@ -190,6 +194,7 @@ abstract class base_room extends base_info
 
         $msg['body']['act']  = 'room';
         $msg['body']['op']   = $op;
+        $msg['body']['msgtype'] = "message";
         $msg['body']['room'] = $this->mRoomid;
         $msg['body']['content'] = $content;
 
@@ -210,6 +215,7 @@ abstract class base_room extends base_info
 
         $msg['body']['act']  = 'room';
         $msg['body']['op']   = $op;
+        $msg['body']['msgtype'] = "message";
         $msg['body']['room'] = $this->mRoomid;
         $msg['body']['content'] = $content;
 

helper/room/factory_client.php

@@ -42,11 +42,11 @@ class factory_client extends tcp_client
         $port = $config['room_factory']['port'];
         return "tcp://{$host}:{$port}";
     }
-
+    //////////////////////////////////////fcgi//////////////////////////////////////////////////////////////////////////
     public function create_bargain($goods_id,$user,$lowest_price,$usable_days,$random = true,$total_num = 10,$finvite = true)
     {
         $random = $random ? 1 : 0;
-        $param = ["act" => 'factory','op' => 'create', "type" => 'bargain_goods','creator' => $user,'invite' => $finvite,
+        $param = ["act" => 'fcgi','op' => 'create', "type" => 'bargain_goods','creator' => $user,'invite' => $finvite,
             'goods_id' => $goods_id,
             'lowest_price' => $lowest_price,
             'usable_days' => $usable_days,
@@ -65,7 +65,7 @@ class factory_client extends tcp_client
 
     public function create_chat($creator,$user,$finvite = true)
     {
-        $param = ["act" => 'factory','op' => 'create', "type" => 'chat','creator' => $creator,'user' => $user,'invite' => $finvite];
+        $param = ["act" => 'fcgi','op' => 'create', "type" => 'chat','creator' => $creator,'user' => $user,'invite' => $finvite];
         $result = $this->request($param);
         if(empty($result)) return false;
 
@@ -80,7 +80,7 @@ class factory_client extends tcp_client
 
     public function create_group($creator,$finvite = true)
     {
-        $param = ["act" => 'factory','op' => 'create', "type" => 'group','creator' => $creator,'invite' => $finvite];
+        $param = ["act" => 'fcgi','op' => 'create', "type" => 'group','creator' => $creator,'invite' => $finvite];
         $result = $this->request($param);
         if(empty($result)) return false;
 
@@ -95,7 +95,7 @@ class factory_client extends tcp_client
 
     public function create_shake($creator,$finvite = true)
     {
-        $param = ["act" => 'factory','op' => 'create', "type" => 'shake_bonus','creator' => $creator,'invite' => $finvite];
+        $param = ["act" => 'fcgi','op' => 'create', "type" => 'shake_bonus','creator' => $creator,'invite' => $finvite];
         $result = $this->request($param);
         if(empty($result)) return false;
 
@@ -108,10 +108,9 @@ class factory_client extends tcp_client
         }
     }
 
-    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
     public function invite($roomid, $inviter,$invitees)
     {
-        $param = ["act" => 'factory','op' => 'invite','room' => $roomid,'inviter' => $inviter,'invitees' => $invitees];
+        $param = ["act" => 'fcgi','op' => 'invite','room' => $roomid,'inviter' => $inviter,'invitees' => $invitees];
         $result = $this->request($param);
         if(empty($result)) return false;
 
@@ -125,7 +124,7 @@ class factory_client extends tcp_client
     }
     public function leave($roomid, $user)
     {
-        $param = ["act" => 'factory','op' => 'invite','room' => $roomid,'user' => $user];
+        $param = ["act" => 'fcgi','op' => 'invite','room' => $roomid,'user' => $user];
         $result = $this->request($param);
         if(empty($result)) return false;
 
@@ -140,7 +139,22 @@ class factory_client extends tcp_client
 
     public function push($content)
     {
-        $param = ["act" => 'factory','op' => 'push','content' => $content];
+        $param = ["act" => 'fcgi','op' => 'push','content' => $content];
+        $result = $this->request($param);
+        if(empty($result)) return false;
+
+        $code = intval($result['code']);
+        if($code != 200) {
+            return false;
+        }
+        else {
+            return true;
+        }
+    }
+    //////////////////////////////////////access////////////////////////////////////////////////////////////////////////
+    public function access()
+    {
+        $param = ["act" => 'access','op' => 'who'];
         $result = $this->request($param);
         if(empty($result)) return false;
 

helper/room/factory_handler.php

@@ -1,22 +0,0 @@
-<?php
-/**
- * Created by PhpStorm.
- * User: stanley-king
- * Date: 2017/12/23
- * Time: 下午5:39
- */
-
-
-function ev_accept($socket, $flag, $base)
-{
-    room\factory_server::instance()->ev_accept($socket,$flag,$base);
-}
-function ev_read($buffer, $id)
-{
-    room\factory_server::instance()->ev_read($buffer,$id);
-}
-
-function ev_error($buffer, $error, $id)
-{
-    room\factory_server::instance()->ev_error($buffer,$error,$id);
-}

helper/room/factory_processor.php

@@ -11,18 +11,23 @@ namespace room;
 use search\IProcessor;
 use errcode;
 use scope_trace;
+use uniquer;
 
 class factory_processor implements IProcessor
 {
+    const room_connection = "room_connection";
+
     private $mFactory;
-    private $mRoomClients;
     private $mLastBuilding;
+    private $mAccUniquer;
+    private $mBufidRooms;
 
     public function __construct()
     {
         $this->mFactory = new factory();
-        $this->mRoomClients = [];
         $this->mLastBuilding = 0;
+        $this->mAccUniquer = new uniquer();
+        $this->mBufidRooms = [];
         $this->init();
     }
 
@@ -31,11 +36,46 @@ class factory_processor implements IProcessor
         global $config;
         $room_addrs = $config['room_factory']['rooms_addr'];
         foreach ($room_addrs as $addr) {
-            $room_client = new room_client($addr['host'],$addr['port']);
-            $this->mRoomClients[] = $room_client;
+            factory_server::instance()->connect($addr['host'],$addr['port'],self::room_connection);
+        }
+    }
+
+    public function onConnected($bufid,$stream,$host,$port,$args)
+    {
+        new scope_trace(__METHOD__);
+
+        if($args == self::room_connection) {
+            $client = new room_client($host,$port,$stream,false);
+            $this->mBufidRooms[$bufid] = $client;
+            $this->block($bufid);
+            $client->init_rooms($bufid);
+            $this->unblock($bufid);
         }
     }
 
+    public function onClose($bufid)
+    {
+        new scope_trace(__METHOD__);
+
+        //if access connection
+        if($this->mAccUniquer->remove_value($bufid)) return;
+
+        //if room connection
+        if(array_key_exists($bufid,$this->mBufidRooms)) {
+            $client = $this->mBufidRooms[$bufid];
+            $addr = $client->host_port();
+            unset($this->mBufidRooms[$bufid]);
+            factory_server::instance()->connect($addr['host'],$addr['port'],self::room_connection);
+        }
+    }
+
+    private function block($bufid) {
+        factory_server::instance()->block($bufid);
+    }
+    private function unblock($bufid) {
+        factory_server::instance()->unblock($bufid);
+    }
+
     public function onRequest($bufid, $body)
     {
         new scope_trace(__METHOD__);
@@ -45,8 +85,15 @@ class factory_processor implements IProcessor
         }
 
         $act = $input['act'];
-        if(!empty($act) && $act == proto_type::act_factory) {
-            $ret = $this->onFactory($input);
+        if(empty($act)) return false;
+
+        if($act == proto_type::act_fcgi) {
+            $ret = $this->onFcgi($bufid,$input);
+            factory_server::instance()->write($bufid,$ret);
+            return true;
+        }
+        elseif($act == proto_type::act_access) {
+            $ret = $this->onAccess($bufid,$input);
             factory_server::instance()->write($bufid,$ret);
             return true;
         }
@@ -55,11 +102,8 @@ class factory_processor implements IProcessor
         }
     }
 
-    public function onClose($bufid)
-    {
-    }
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-    private function onFactory($input)
+    private function onFcgi($bufid, $input)
     {
         $op = $input['op'];
         if($op == 'create')
@@ -140,24 +184,41 @@ class factory_processor implements IProcessor
             return $this->error(errcode::ErrRoomFactoryOp);
         }
     }
-
+    private function onAccess($bufid,$input)
+    {
+        $op = $input['op'];
+        if($op == 'who') {
+            $this->mAccUniquer->add_value($bufid);
+            return $this->success(['act' => proto_type::act_access,'op' => 'who']);
+        }
+    }
+    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
     private function invite($roomid, $inviter,$invitees)
     {
-        foreach ($this->mRoomClients as  $client)
+        foreach ($this->mBufidRooms as $bufid => $client)
         {
             if($client->contain_room($roomid))
             {
+                $this->block($bufid);
                 $ret = $client->invite($roomid,$inviter,$invitees);
+                $this->unblock($bufid);
+
                 if($ret != false) {
-                    return $ret;
+                    //todo broad cast all access
                 }
+
+                return $ret;
             }
         }
 
-        $client = $this->room_client();
-        if($client != false)
+        $bufid = $this->room_bufid();
+        if($bufid != false)
         {
+            $client = $this->mBufidRooms[$bufid];
+            $this->block($bufid);
             $ret = $client->invite($roomid,$inviter,$invitees);
+            $this->unblock($bufid);
+
             if($ret != false) {
                 $client->add_room($roomid);
             }
@@ -168,57 +229,68 @@ class factory_processor implements IProcessor
 
     private function leave($roomid, $user)
     {
-        foreach ($this->mRoomClients as  $client)
+        foreach ($this->mBufidRooms as $bufid => $client)
         {
             if($client->contain_room($roomid))
             {
+                $this->block($bufid);
                 $ret = $client->leave($roomid,$user);
+                $this->unblock($bufid);
+
                 if($ret != false) {
-                    return $ret;
+                    //todo broad cast all access
                 }
+                return $ret;
             }
         }
-
-        $client = $this->room_client();
-        if($client != false) {
-            return $client->leave($roomid,$user);
-        } else {
-            return false;
-        }
     }
 
     private function push($content)
     {
-        foreach ($this->mRoomClients as  $client)
+        $bufid = $this->room_bufid();
+        if($bufid != false)
         {
+            $client = $this->mBufidRooms[$bufid];
+            $this->block($bufid);
             $ret = $client->push($content);
-            if($ret != false) {
-                return true;
-            }
+            $this->unblock($bufid);
+            return true;
         }
+
         return false;
     }
 
-    private function room_client()
+    private function room_bufid()
     {
-        $count = count($this->mRoomClients);
+        $count = count($this->mBufidRooms);
         if($count <= 0) return false;
 
-        $pos = $this->mLastBuilding;
-        if($pos < $count) {
-            $this->mLastBuilding++;
-            return $this->mRoomClients[$pos];
-        } else {
+        if($this->mLastBuilding >= $count) {
             $this->mLastBuilding = 0;
-            return $this->mRoomClients[0];
         }
+
+        $pos = 0;
+        foreach ($this->mBufidRooms as $bufid => $client)
+        {
+            if($pos != $this->mLastBuilding) {
+                $pos++;
+                continue;
+            }
+            else {
+                break;
+            }
+        }
+
+        return $bufid;
     }
 
-    private function success($datas) {
+    private function success($datas)
+    {
         $code = errcode::Success;
         $data['code'] = $code;
         $data['message'] = errcode::msg($code);
         $data['data'] = $datas;
+        $data['msgtype'] = "reply";
 
         return json_encode($data);
     }
@@ -229,10 +301,11 @@ class factory_processor implements IProcessor
             $message = errcode::msg($code);
         }
 
-        $data = array();
         $data['code'] = $code;
         $data['message'] = $message;
         $data['datas'] = null;
+        $data['msgtype'] = "reply";
+
         return json_encode($data);
     }
 }

helper/room/proto_type.php

@@ -12,9 +12,11 @@ namespace room;
 class proto_type
 {
     //消息类型
+    const act_fcgi    = 'fcgi';
     const act_factory = 'factory';
     const act_room    = 'room';
     const act_access  = 'access';
+
     const act_chatwo  = 'chatwo';
     const act_push    = 'push';
 

helper/room/room_client.php

@@ -18,16 +18,17 @@ class room_client extends tcp_client
     private $mHost;
     private $mPort;
     private $mRooms;
+    private $mBufid;
 
-    public function __construct($host,$port)
+    public function __construct($host,$port,$socket = false,$auto_con = true)
     {
         $this->mHost = $host;
         $this->mPort = $port;
         $this->mRooms = [];
+        $this->mBufid = 0;
 
-        parent::__construct();
+        parent::__construct($socket,$auto_con);
         $this->mBodyType = tcp_client::JsonType;
-        $this->init();
     }
 
     public function __destruct()
@@ -35,8 +36,13 @@ class room_client extends tcp_client
         parent::__destruct();
     }
 
-    public function init()
+    public function buffer_id() {
+        return $this->mBufid;
+    }
+
+    public function init_rooms($bufid)
     {
+        $this->mBufid = $bufid;
         $ret = $this->list_room();
         if($ret != false)
         {
@@ -80,6 +86,9 @@ class room_client extends tcp_client
     public function remote_addr() {
         return "tcp://{$this->mHost}:{$this->mPort}";
     }
+    public function host_port() {
+        return ['host' => $this->mHost,'port' => $this->mPort];
+    }
 
     public function invite($roomid, $inviter,$invitees)
     {

helper/room/room_handler.php

@@ -1,21 +0,0 @@
-<?php
-/**
- * Created by PhpStorm.
- * User: stanley-king
- * Date: 2017/12/14
- * Time: 下午5:15
- */
-
-function ev_accept($socket, $flag, $base)
-{
-    room\room_server::instance()->ev_accept($socket,$flag,$base);
-}
-function ev_read($buffer, $id)
-{
-    room\room_server::instance()->ev_read($buffer,$id);
-}
-
-function ev_error($buffer, $error, $id)
-{
-    room\room_server::instance()->ev_error($buffer,$error,$id);
-}

helper/room/room_processor.php

@@ -28,6 +28,11 @@ class room_processor implements IProcessor
         $this->mChatwo = new chatwo();
     }
 
+    public function onConnected($bufid,$stream,$host,$port,$args)
+    {
+
+    }
+
     public function onRequest($bufid, $body)
     {
         new scope_trace(__METHOD__);
@@ -206,6 +211,7 @@ class room_processor implements IProcessor
                 if($success) {
                     $this->write_roomsg($bufid,$room);
                 }
+
                 return true;
             }
         }

helper/search/event_handler.php

@@ -1,21 +0,0 @@
-<?php
-/**
- * Created by PhpStorm.
- * User: stanley-king
- * Date: 2016/10/15
- * Time: 下午4:41
- */
-
-function ev_accept($socket, $flag, $base)
-{
-    search\CenterHelper::instance()->ev_accept($socket,$flag,$base);
-}
-function ev_read($buffer, $id)
-{
-    search\CenterHelper::instance()->ev_read($buffer,$id);
-}
-
-function ev_error($buffer, $error, $id)
-{
-    search\CenterHelper::instance()->ev_error($buffer,$error,$id);
-}

helper/search/processor.php

@@ -26,6 +26,10 @@ class processor implements IProcessor
 
     const ValidateArea   = 20;
 
+    public function onConnected($bufid,$stream,$host,$port,$args)
+    {
+
+    }
     public function onRequest($bufid, $body)
     {
         $checker = new scope_trace(__METHOD__);

helper/search/srv_base.php

@@ -15,6 +15,7 @@ interface IProcessor
 {
     public function onRequest($bufid, $body);
     public function onClose($bufid);
+    public function onConnected($bufid,$stream,$host,$port,$args);
 }
 
 abstract class srv_base
@@ -23,20 +24,29 @@ abstract class srv_base
     const read_time_out  = 600;
     const write_time_out = 5;
 
-    private $mListenSocket;
-    protected $mEvbase;
-    protected $mEv;
+    protected $mEvBase;
+    protected $mEvAccepts;
+    protected $mEvSignals;
+    protected $mEvConnects;
 
     protected $mStreams;
     protected $mBuffers;
     protected $mContents;
     protected $mProcessor;
+    protected $mBufferID;
 
     protected function __construct()
     {
         $this->mStreams  = [];
         $this->mBuffers  = [];
         $this->mContents = [];
+        $this->mEvAccepts = [];
+        $this->mEvSignals = [];
+        $this->mEvConnects = [];
+        $this->mEvTimeouts = [];
+
+        $this->mBufferID = 1;
+        $this->mEvBase   = event_base_new();
     }
 
     public function init(IProcessor $processor)
@@ -44,71 +54,246 @@ abstract class srv_base
         $this->mProcessor = $processor;
     }
 
-    public function run_loop($sockfd)
+    public function run_loop($sock)
     {
-        $this->mListenSocket = $sockfd;
-        $this->mEvbase = event_base_new();
-        $this->mEv = event_new();
+        $this->add_listen($sock);
+        $this->add_signal(SIGINT);
+        $this->add_signal(SIGQUIT);
+        $this->add_signal(SIGTERM);
+
+        $ret = event_base_loop($this->mEvBase);
+        Log::record("event_base_loop ret={$ret}",Log::DEBUG);
+    }
 
-        Log::record("event_set",Log::DEBUG);
-        if(event_set($this->mEv, $this->mListenSocket, EV_READ | EV_PERSIST, 'ev_accept', $this->mEvbase) == false) {
+    private function add_listen($fd)
+    {
+        $fd = intval($fd);
+        if($fd < 0) return false;
+
+        $this->mEvAccepts[$fd] = event_new();
+        if(event_set($this->mEvAccepts[$fd], $fd, EV_READ | EV_PERSIST, [$this, 'onAccept'], $this->mEvBase) == false) {
             Log::record("event_set error EV_READ | EV_PERSIST",Log::DEBUG);
         }
 
-        Log::record("event_base_set",Log::DEBUG);
-        if(event_base_set($this->mEv, $this->mEvbase) == false) {
+        if(event_base_set($this->mEvAccepts[$fd], $this->mEvBase) == false) {
             Log::record("event_set error EV_READ | EV_PERSIST",Log::DEBUG);
         }
 
-        Log::record("event_add",Log::DEBUG);
-        if(event_add($this->mEv) == false) {
+        if(event_add($this->mEvAccepts[$fd]) == false) {
             Log::record("event_add error EV_READ | EV_PERSIST",Log::DEBUG);
         }
+    }
 
-        $ret = event_base_loop($this->mEvbase);
-        Log::record("event_base_loop ret={$ret}",Log::DEBUG);
+    private function add_signal($val)
+    {
+        $this->mEvSignals[$val] = event_new();
+
+        if(event_set($this->mEvSignals[$val], $val, EV_SIGNAL, [$this, 'onSignal']) == false) {
+            Log::record("event_set error EV_SIGNAL sig={$val}",Log::DEBUG);
+        }
+
+        if(event_base_set($this->mEvSignals[$val], $this->mEvBase) == false) {
+            Log::record("event_base_set error EV_SIGNAL sig={$val}",Log::DEBUG);
+        }
+
+        if(event_add($this->mEvSignals[$val]) == false) {
+            Log::record("event_add error EV_SIGNAL sig={$val}",Log::DEBUG);
+        }
     }
 
-    public function ev_accept($socket, $flag, $base)
+    public function onSignal($sig, $flag)
     {
-        $pid = posix_getpid();
-        Log::record("ev_accept pid={$pid} socket_fd={$socket}",Log::DEBUG);
+        Log::record("ev_signal sig={$sig},flag={$flag}",Log::DEBUG);
+        if($sig == SIGINT || $sig == SIGQUIT || $sig == SIGTERM)
+        {
+            event_base_loopexit($this->mEvBase);
 
-        static $bufid = 1;
+            foreach ($this->mBuffers as $bufid => $buffer) {
+                event_buffer_disable($buffer, EV_READ | EV_WRITE);
+                event_buffer_free($buffer);
 
+                @socket_shutdown($this->mStreams[$bufid],STREAM_SHUT_RDWR);
+                socket_close($this->mStreams[$bufid]);
+            }
+            $this->mStreams = [];
+            $this->mContents = [];
+            $this->mBuffers = [];
+        }
+    }
+    public function onAccept($socket, $event, $base)
+    {
+        Log::record("onAccept socket_fd={$socket} flag={$event}",Log::DEBUG);
         $stream = socket_accept($socket);
         if($stream == false) {
-            Log::record("stream_socket_accept return false pid={$pid} socket_fd={$socket}",Log::DEBUG);
+            Log::record("stream_socket_accept return false socket_fd={$socket}",Log::DEBUG);
             return;
         }
-        Log::record("stream_socket_accept pid={$pid}  stream={$stream}",Log::DEBUG);
+        Log::record("stream_socket_accept stream={$stream}",Log::DEBUG);
+        $this->add_stream($stream,$base);
+    }
 
+    private function add_stream($stream,$base)
+    {
+        $bufid = $this->mBufferID;
         socket_set_nonblock($stream);
-        $buffer = event_buffer_new($stream, 'ev_read', NULL, 'ev_error', $bufid);
+        $buffer = event_buffer_new($stream, [$this,'onRead'], NULL, [$this,'onError'], $bufid);
+
         if($buffer == false) {
-            fclose($stream);
-            Log::record("event_buffer_new return false pid={$pid} socket_fd={$socket}",Log::DEBUG);
-            return;
+            socket_close($stream);
+            return false;
         }
+        else
+        {
+            event_buffer_base_set($buffer, $base);
+            event_buffer_timeout_set($buffer, self::read_time_out, self::write_time_out);
+            event_buffer_watermark_set($buffer, EV_READ, 0, 0xffffff);
+            event_buffer_priority_set($buffer, 10);
+            event_buffer_enable($buffer, EV_READ | EV_PERSIST);
+
+            $this->mStreams[$bufid] = $stream;
+            $this->mBuffers[$bufid] = $buffer;
+            $this->mContents[$bufid] = "";
+
+            $this->mBufferID++;
+            if($this->mBufferID < 0) $this->mBufferID = 1;
+
+            return $bufid;
+        }
+    }
+
+    public function block($bufid)
+    {
+        $stream = $this->mStreams[$bufid];
+        $buffer = $this->mBuffers[$bufid];
+
+        $success = event_buffer_disable($buffer, EV_READ | EV_PERSIST);
+        Log::record("event_buffer_disable success={$success}",Log::DEBUG);
+
+        $success = socket_set_block($stream);
+        Log::record("socket_set_block success={$success}",Log::DEBUG);
+    }
 
-        event_buffer_base_set($buffer, $base);
-        event_buffer_timeout_set($buffer, self::read_time_out, self::write_time_out);
-        event_buffer_watermark_set($buffer, EV_READ, 0, 0xffffff);
-        event_buffer_priority_set($buffer, 10);
-        event_buffer_enable($buffer, EV_READ | EV_PERSIST);
+    public function unblock($bufid)
+    {
+        $stream = $this->mStreams[$bufid];
+        $buffer = $this->mBuffers[$bufid];
 
-        $this->mStreams[$bufid] = $stream;
-        $this->mBuffers[$bufid] = $buffer;
-        $this->mContents[$bufid] = "";
+        $success = socket_set_nonblock($stream);
+        Log::record("socket_set_nonblock success={$success}",Log::DEBUG);
 
-        $bufid++;
-        if($bufid < 0) $bufid = 1;
+        $success = event_buffer_enable($buffer, EV_READ | EV_PERSIST);
+        Log::record("event_buffer_enable success={$success}",Log::DEBUG);
     }
 
-    public function ev_read($buffer, $bufid)
+    public function onConnectTimer($stream, $event, $params)
     {
-        $pid = posix_getpid();
-        Log::record("ev_read begin pid={$pid}",Log::DEBUG);
+        $error = socket_last_error();
+        Log::record("onConnectTimer sig={$stream},flag={$event},error={$error}",Log::DEBUG);
+        $stream = $this->do_connect($params['host'],$params['port'],$params['args'],$params['stream']);
+        if($stream != false) {
+            $bufid = $this->add_stream($stream,$this->mEvBase);
+            $this->mProcessor->onConnected($bufid, $stream,$params['host'],$params['port'],$params['args']);
+        }
+    }
+
+    public function onConnect($stream, $event, $params)
+    {
+        $error = socket_last_error();
+        Log::record("onConnect sig={$stream},flag={$event},socket error={$error}",Log::DEBUG);
+
+        if($event == EV_WRITE) {
+            $ret = socket_getopt($stream,SOL_SOCKET,SO_ERROR);
+            if($ret == 0) {
+                Log::record("onConnect EV_WRITE success ",Log::DEBUG);
+                $fd = intval($stream);
+                event_del($this->mEvConnects[$fd]);
+                event_free($this->mEvConnects[$fd]);
+                event_del($this->mEvTimeouts[$fd]);
+                event_free($this->mEvTimeouts[$fd]);
+                unset($this->mEvConnects[$fd]);
+                unset($this->mEvTimeouts[$fd]);
+                $bufid = $this->add_stream($stream, $this->mEvBase);
+                $this->mProcessor->onConnected($bufid, $stream,$params['host'],$params['port'],$params['args']);
+            }
+            else {
+                Log::record("onConnect EV_WRITE Error",Log::DEBUG);
+            }
+        }
+        elseif($event == EV_WRITE | EV_READ) {
+            Log::record("onConnect EV_WRITE | EV_READ",Log::DEBUG);
+        }
+        else {
+            Log::record("onConnect ETIMEOUT",Log::DEBUG);
+        }
+    }
+
+    private function do_connect($host,$port,$args,$old_stream = false)
+    {
+        $stream = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
+        socket_set_nonblock($stream);
+        $fd = intval($stream);
+
+        if($old_stream == false)
+        {
+            $this->mEvConnects[$fd] = event_new();
+            $this->mEvTimeouts[$fd] = event_timer_new();
+        }
+        else {
+            $oldfd = intval($old_stream);
+            $this->mEvConnects[$fd] = $this->mEvConnects[$oldfd];
+            $this->mEvTimeouts[$fd] = $this->mEvTimeouts[$oldfd];
+            unset($this->mEvConnects[$oldfd]);
+            unset($this->mEvTimeouts[$oldfd]);
+        }
+
+        $ret = socket_connect($stream,$host,$port);
+        if($ret == false)
+        {
+            if(event_set($this->mEvConnects[$fd], $stream, EV_WRITE | EV_READ, [$this, 'onConnect'],['host' => $host,'port' => $port,'args' => $args]) == false) {
+                Log::record("event_set error connect fd={$stream}",Log::DEBUG);
+            }
+            if(event_base_set($this->mEvConnects[$fd], $this->mEvBase) == false) {
+                Log::record("event_base_set error connect fd={$stream}",Log::DEBUG);
+            }
+            if(event_timer_set($this->mEvTimeouts[$fd], [$this, 'onConnectTimer'],['stream' => $stream, 'host' => $host,'port' => $port,'args' => $args]) == false) {
+                Log::record("event_set error connect fd={$stream}",Log::DEBUG);
+            }
+            if(event_base_set($this->mEvTimeouts[$fd], $this->mEvBase) == false) {
+                Log::record("event_base_set error connect fd={$stream}",Log::DEBUG);
+            }
+            if(event_add($this->mEvConnects[$fd]) == false) {
+                Log::record("event_add error connect fd={$stream}",Log::DEBUG);
+            }
+
+            if(event_timer_add($this->mEvTimeouts[$fd],1000000) == false) {
+                Log::record("event_base_set error connect fd={$stream}",Log::DEBUG);
+            }
+            return false;
+        }
+        else {
+            event_del($this->mEvConnects[$fd]);
+            event_free($this->mEvConnects[$fd]);
+            event_del($this->mEvTimeouts[$fd]);
+            event_free($this->mEvTimeouts[$fd]);
+
+            unset($this->mEvConnects[$fd]);
+            unset($this->mEvTimeouts[$fd]);
+
+            return $stream;
+        }
+    }
+    public function connect($host,$port,$args)
+    {
+        $stream = $this->do_connect($host,$port,$args,false);
+        if($stream != false) {
+            $bufid = $this->add_stream($stream,$this->mEvBase);
+            $this->mProcessor->onConnected($bufid, $stream,$host,$port,$args);
+        }
+    }
+
+    public function onRead($buffer, $bufid)
+    {
+        Log::record("onRead bufid={$bufid}",Log::DEBUG);
 
         $content = &$this->mContents[$bufid];
         while (true)
@@ -154,32 +339,37 @@ abstract class srv_base
         if($start > 0) {
             $content = substr($content,$start);
         }
-        Log::record("ev_read end pid={$pid}",Log::DEBUG);
     }
 
-    public function ev_error($buffer, $error, $bufid)
+    public function onError($buffer, $event, $bufid)
     {
-        $error = socket_strerror($error);
-        Log::record("ev_error id={$bufid} error={$error}",Log::DEBUG);
-
-        event_buffer_disable($buffer, EV_READ | EV_WRITE);
-        event_buffer_free($buffer);
+        Log::record("onError bufid={$bufid} flag={$event}",Log::DEBUG);
 
-        if(array_key_exists($bufid,$this->mStreams)) {
-            socket_shutdown($this->mStreams[$bufid],STREAM_SHUT_RDWR);
-            fclose($this->mStreams[$bufid]);
-            unset($this->mStreams[$bufid]);
+        if ($event == (EVBUFFER_READ | EVBUFFER_TIMEOUT)) {
+            event_buffer_enable($buffer, EV_READ);
+            $this->write($bufid,'');
         }
+        else
+        {
+            event_buffer_disable($buffer, EV_READ | EV_WRITE);
+            event_buffer_free($buffer);
 
-        if(array_key_exists($bufid,$this->mBuffers)) {
-            unset($this->mBuffers[$bufid]);
-        }
-        if(array_key_exists($bufid,$this->mContents)) {
-            unset($this->mContents[$bufid]);
-        }
+            if(array_key_exists($bufid,$this->mStreams)) {
+                @socket_shutdown($this->mStreams[$bufid],STREAM_SHUT_RDWR);
+                socket_close($this->mStreams[$bufid]);
+                unset($this->mStreams[$bufid]);
+            }
 
-        if($this->mProcessor != null) {
-            $this->mProcessor->onClose($bufid);
+            if(array_key_exists($bufid,$this->mBuffers)) {
+                unset($this->mBuffers[$bufid]);
+            }
+            if(array_key_exists($bufid,$this->mContents)) {
+                unset($this->mContents[$bufid]);
+            }
+
+            if($this->mProcessor != null) {
+                $this->mProcessor->onClose($bufid);
+            }
         }
     }
 
@@ -194,6 +384,7 @@ abstract class srv_base
             return false;
         }
     }
+
     public function close($bufid)
     {
         if(!array_key_exists($bufid,$this->mBuffers)) return false;
@@ -203,8 +394,8 @@ abstract class srv_base
         event_buffer_free($buffer);
 
         if(array_key_exists($bufid,$this->mStreams)) {
-            socket_shutdown($this->mStreams[$bufid],STREAM_SHUT_RDWR);
-            fclose($this->mStreams[$bufid]);
+            @socket_shutdown($this->mStreams[$bufid],STREAM_SHUT_RDWR);
+            socket_close($this->mStreams[$bufid]);
             unset($this->mStreams[$bufid]);
         }
 

helper/search/tcp_client.php

@@ -18,13 +18,16 @@ abstract class tcp_client
     //协议打包格式，PHP序列化用于PHP进程间通信，json 格式用于PHP与C++进程间通信
     const SerializeType = 1;
     const JsonType  = 2;
-    private $mSocket;
 
+    private $mSocket;
     protected $mBodyType;
 
-    protected function __construct()
+    private $mAutoConnect;
+
+    protected function __construct($socket = false,$auto_connect = true)
     {
-        $this->mSocket = false;
+        $this->mSocket = $socket;
+        $this->mAutoConnect = $auto_connect;
     }
 
     public function __destruct()
@@ -70,6 +73,8 @@ abstract class tcp_client
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
     private function init_socket()
     {
+        if(!$this->mAutoConnect) return true;
+
         if($this->mSocket != false)
         {
             if(feof($this->mSocket) == true) {
@@ -93,6 +98,8 @@ abstract class tcp_client
 
     private function fini_socket()
     {
+        if(!$this->mAutoConnect) return;
+
         if($this->mSocket != false) {
             Log::record("tcp_client fini_socket",Log::DEBUG);
             stream_socket_shutdown($this->mSocket,STREAM_SHUT_RDWR);
@@ -103,8 +110,14 @@ abstract class tcp_client
 
     private function read_body()
     {
-        $body_len = $this->read(self::body_header_len);
-        $body_len = intval($body_len);
+        do {
+
+            $body = $this->read(self::body_header_len);
+            if($body === false) return false;
+            $body_len = intval($body);
+        }
+        while($body_len === 0);
+
         if($body_len > 0) {
             $body = $this->read($body_len);
             return $body;
@@ -119,13 +132,21 @@ abstract class tcp_client
         $data = "";
         while ($left > 0)
         {
-            if(feof($this->mSocket) == true) {
-                return false;
-            } else {
-                $tmp = fread($this->mSocket,$left);
-                $left = $left - strlen($tmp);
-                $data .= $tmp;
+            if($this->mAutoConnect)
+            {
+                if (feof($this->mSocket) == true) {
+                    return false;
+                } else {
+                    $tmp = fread($this->mSocket, $left);
+                }
             }
+            else {
+                $tmp = @socket_read($this->mSocket,$left);
+                if(strlen($tmp) == 0) return false;
+            }
+
+            $left = $left - strlen($tmp);
+            $data .= $tmp;
         }
 
         return $data;
@@ -134,7 +155,11 @@ abstract class tcp_client
     {
         while ($len > 0)
         {
-            $ret = fwrite($this->mSocket,$data);
+            if($this->mAutoConnect) {
+                $ret = fwrite($this->mSocket,$data);
+            } else {
+                $ret = @socket_write($this->mSocket,$data);
+            }
             if($ret == false || $ret <= 0) {
                 return false;
             }

mobile/control/member_talk.php

@@ -31,4 +31,40 @@ class member_talkControl extends mobileControl //mbMemberControl
         $result['token'] = room\author::sign_native($userid);
         return self::outsuccess($result);
     }
+
+    public function authon_webOp()
+    {
+        global $config;
+        $room_id = $config['special_rooms']['shake_bonus'];
+        $webaddr = $config['access_addr'];
+        $userid = intval($_GET['userid']);
+
+        $creator = self::shake_creator($room_id);
+        $ret = room\factory_client::instance()->invite($room_id,$creator,[$userid]);
+
+        $result = [];
+        if($ret != false) {
+            $result['addr'] = $webaddr;
+            $result['token'] = room\author::sign_web($room_id,session_helper::memberid());
+            $result['room'] = intval($room_id);
+            $result['user'] = session_helper::memberid();
+        }
+
+        $_SESSION['client_type'] = "wap";
+        return self::outsuccess($result);
+    }
+
+    static private function shake_creator($room_id)
+    {
+        static $stCreator = 0;
+
+        if($stCreator == 0) {
+            $mod_room = Model('room');
+            $params = $mod_room->getRoom($room_id);
+            $rinfo = new room\base_info($params);
+            $stCreator = $rinfo->creator();
+        }
+
+        return $stCreator;
+    }
 }

room_factory.php

@@ -12,7 +12,6 @@ require_once(BASE_ROOT_PATH . '/fooder.php');
 require_once(BASE_ROOT_PATH . '/helper/search/srv_base.php');
 require_once(BASE_ROOT_PATH . '/helper/room/factory_server.php');
 require_once(BASE_ROOT_PATH . '/helper/room/factory_processor.php');
-require_once(BASE_ROOT_PATH . '/helper/room/factory_handler.php');
 require_once(BASE_ROOT_PATH . '/helper/room/proto_type.php');
 require_once(BASE_ROOT_PATH . '/helper/room/base_room.php');
 require_once(BASE_ROOT_PATH . '/helper/room/chat_room.php');

room_srv.php

@@ -12,7 +12,6 @@ require_once(BASE_ROOT_PATH . '/fooder.php');
 require_once(BASE_ROOT_PATH . '/helper/search/srv_base.php');
 require_once(BASE_ROOT_PATH . '/helper/room/room_server.php');
 require_once(BASE_ROOT_PATH . '/helper/room/room_processor.php');
-require_once(BASE_ROOT_PATH . '/helper/room/room_handler.php');
 require_once(BASE_ROOT_PATH . '/helper/room/proto_type.php');
 require_once(BASE_ROOT_PATH . '/helper/room/base_room.php');
 require_once(BASE_ROOT_PATH . '/helper/room/chat_room.php');

test/TestRoomAccess.php

@@ -10,7 +10,6 @@ require_once(BASE_ROOT_PATH . '/fooder.php');
 require_once(BASE_ROOT_PATH . '/helper/search/srv_base.php');
 require_once(BASE_ROOT_PATH . '/helper/room/factory_server.php');
 require_once(BASE_ROOT_PATH . '/helper/room/factory_processor.php');
-require_once(BASE_ROOT_PATH . '/helper/room/event_handler.php');
 require_once(BASE_ROOT_PATH . '/helper/room/proto_type.php');
 require_once(BASE_ROOT_PATH . '/helper/room/factory.php');
 require_once(BASE_ROOT_PATH . '/helper/room/croom.php');

test/TestRoomFactory.php

@@ -13,7 +13,6 @@ require_once(BASE_ROOT_PATH . '/helper/search/tcp_client.php');
 
 require_once(BASE_ROOT_PATH . '/helper/room/factory_server.php');
 require_once(BASE_ROOT_PATH . '/helper/room/factory_processor.php');
-require_once(BASE_ROOT_PATH . '/helper/room/factory_handler.php');
 require_once(BASE_ROOT_PATH . '/helper/room/proto_type.php');
 require_once(BASE_ROOT_PATH . '/helper/room/base_room.php');
 require_once(BASE_ROOT_PATH . '/helper/room/chat_room.php');
@@ -24,7 +23,7 @@ require_once(BASE_ROOT_PATH . '/helper/room/factory.php');
 require_once(BASE_ROOT_PATH . '/helper/room/room_client.php');
 require_once(BASE_ROOT_PATH . '/helper/room/factory_client.php');
 
-
+use room;
 class TestRoomFactory extends PHPUnit_Framework_TestCase
 {
     const admin_member_id = 36429;
@@ -41,7 +40,7 @@ class TestRoomFactory extends PHPUnit_Framework_TestCase
     {
         $processor = new room\factory_processor();
         $creator = 36429;
-        $ret = $processor->onRequest(0,json_encode(["act" => 'factory','op' => 'create', "type" => 'group','creator' => $creator,'invite' => true]));
+        $ret = $processor->onRequest(0,json_encode(["act" => room\proto_type::act_fcgi,'op' => 'create', "type" => 'group','creator' => $creator,'invite' => true]));
     }
 
     public function testProcessorInvite()
@@ -49,22 +48,22 @@ class TestRoomFactory extends PHPUnit_Framework_TestCase
         $processor = new room\factory_processor();
         $roomid=30;
         $user = 36429;
-        $ret = $processor->onRequest(0,json_encode(["act" => 'factory','op' => 'invite','room' => $roomid,'inviter' => $user,'invitees' => [36500]]));
+        $ret = $processor->onRequest(0,json_encode(["act" => room\proto_type::act_fcgi,'op' => 'invite','room' => $roomid,'inviter' => $user,'invitees' => [36500]]));
         $user = 36430;
-        $ret = $processor->onRequest(0,json_encode(["act" => 'factory','op' => 'invite','room' => $roomid,'inviter' => $user,'invitees' => [36500]]));
+        $ret = $processor->onRequest(0,json_encode(["act" => room\proto_type::act_fcgi,'op' => 'invite','room' => $roomid,'inviter' => $user,'invitees' => [36500]]));
     }
     public function testProcessorCreateShake()
     {
         $processor = new room\factory_processor();
         $creator = 36429;
-        $ret = $processor->onRequest(0,json_encode(["act" => 'factory','op' => 'create', "type" => 'shake_bonus','creator' => $creator]));
+        $ret = $processor->onRequest(0,json_encode(["act" => room\proto_type::act_fcgi,'op' => 'create', "type" => 'shake_bonus','creator' => $creator]));
     }
 
     public function testFactoryCreateBargain()
     {
         $factory = new room\factory();
         $creator = 36477;
-        $parms = ["act" => 'factory','op' => 'create', "type" => 'bargain_goods','creator' => $creator,'goods_id' => 4831,'lowest_price' => 10,'usable_days' => 3,'random' => 1,'total_num' => 10];
+        $parms = ["act" => room\proto_type::act_fcgi,'op' => 'create', "type" => 'bargain_goods','creator' => $creator,'goods_id' => 4831,'lowest_price' => 10,'usable_days' => 3,'random' => 1,'total_num' => 10];
         $factory->create($parms);
     }
     public function testFactoryBuildBargain()
@@ -77,14 +76,21 @@ class TestRoomFactory extends PHPUnit_Framework_TestCase
     {
         $processor = new room\factory_processor();
         $creator = 36483;
-        $ret = $processor->onRequest(0,json_encode(["act" => 'factory','op' => 'create', "type" => 'bargain_goods','creator' => $creator,
+        $ret = $processor->onRequest(0,json_encode(["act" => room\proto_type::act_fcgi,'op' => 'create', "type" => 'bargain_goods','creator' => $creator,
             'goods_id' => 4831,'lowest_price' => 10,'usable_days' => 3,'random' => 1,'total_num' => 10]));
     }
 
     public function testInvite()
     {
         $processor = new room\factory_processor();
-        $req = ["act" => 'factory','op' => 'invite','room' => 30,'inviter' => $this->creator(30),'invitees' => [36500]];
+        $req = ["act" => room\proto_type::act_fcgi,'op' => 'invite','room' => 30,'inviter' => $this->creator(30),'invitees' => [36500]];
+        $ret = $processor->onRequest(0,json_encode($req));
+    }
+
+    public function testAccessWho()
+    {
+        $processor = new room\factory_processor();
+        $req = ["act" => room\proto_type::act_access,'op' => 'who'];
         $ret = $processor->onRequest(0,json_encode($req));
     }
 
@@ -104,6 +110,11 @@ class TestRoomFactory extends PHPUnit_Framework_TestCase
         $ret = room\factory_client::instance()->push('hello world');
     }
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+    public function testAccess()
+    {
+        $ret = room\factory_client::instance()->push('hello world');
+    }
+    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
     private function creator($room_id)
     {
         $mod_room = Model('room');

test/TestRoomSrv.php

@@ -11,7 +11,6 @@ require_once(BASE_ROOT_PATH . '/fooder.php');
 require_once(BASE_ROOT_PATH . '/helper/search/srv_base.php');
 require_once(BASE_ROOT_PATH . '/helper/room/room_server.php');
 require_once(BASE_ROOT_PATH . '/helper/room/room_processor.php');
-require_once(BASE_ROOT_PATH . '/helper/room/room_handler.php');
 require_once(BASE_ROOT_PATH . '/helper/room/proto_type.php');
 require_once(BASE_ROOT_PATH . '/helper/room/base_room.php');
 require_once(BASE_ROOT_PATH . '/helper/room/chat_room.php');

test/TestTalk.php

@@ -0,0 +1,58 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: stanley-king
+ * Date: 2018/7/16
+ * Time: 下午4:36
+ */
+define('BASE_ROOT_PATH',str_replace('/test','',dirname(__FILE__)));
+
+require_once (BASE_ROOT_PATH . '/fooder.php');
+require_once(BASE_ROOT_PATH . '/helper/search/tcp_client.php');
+require_once(BASE_ROOT_PATH . '/helper/room/factory_client.php');
+require_once(BASE_ROOT_PATH . '/helper/room/proto_type.php');
+require_once(BASE_ROOT_PATH . '/helper/room/author.php');
+
+class TestTalk extends PHPUnit_Framework_TestCase
+{
+    public static function setUpBeforeClass()
+    {
+        Base::run_util();
+    }
+
+    public function testGen()
+    {
+        global $config;
+        $room_id = $config['special_rooms']['shake_bonus'];
+        $webaddr = $config['access_addr'];
+
+        $uids = [39625,39653];
+        $creator = self::shake_creator($room_id);
+        $ret = room\factory_client::instance()->invite($room_id,$creator,$uids);
+
+        foreach ($uids as $uid)
+        {
+            $result = [];
+            if($ret != false) {
+                $result['addr'] = $webaddr;
+                $result['token'] = room\author::sign_web($room_id,$uid);
+                $result['room'] = intval($room_id);
+                $result['user'] = $uid;
+            }
+        }
+    }
+
+    static private function shake_creator($room_id)
+    {
+        static $stCreator = 0;
+
+        if($stCreator == 0) {
+            $mod_room = Model('room');
+            $params = $mod_room->getRoom($room_id);
+            $rinfo = new room\base_info($params);
+            $stCreator = $rinfo->creator();
+        }
+
+        return $stCreator;
+    }
+}