add public reconnection and debug

zhouaini528 · Jan 14, 2021 · ce2d4f5 · ce2d4f5
1 parent 3500ca6
commit ce2d4f5
Show file tree

Hide file tree

Showing 4 changed files with 73 additions and 13 deletions.
diff --git a/src/Api/WebSocket/SocketClient.php b/src/Api/WebSocket/SocketClient.php
@@ -39,6 +39,8 @@ protected function init(){
         $this->add('del_sub',[]);//正在删除的频道
 
         $this->add('keysecret',[]);//目前总共key
+
+        $this->add('debug',[]);
     }
 
     function keysecret(array $keysecret=[]){
@@ -179,4 +181,10 @@ function test2(){
             echo json_encode($this->client->$v).PHP_EOL;
         }
     }
+
+    function test_reconnection(){
+        $this->client->debug=[
+            'public'=>['public'=>'close'],
+        ];
+    }
 }
diff --git a/src/Api/WebSocket/SocketFunction.php b/src/Api/WebSocket/SocketFunction.php
@@ -100,4 +100,22 @@ protected function log($message){
     protected function userKey(array $keysecret,string $sub){
         return $keysecret['key'].self::$USER_DELIMITER.$sub;
     }
+
+    /**
+     * 重新订阅
+     */
+    private function reconnection($global,$type='public'){
+        $all_sub=$global->get('all_sub');
+        if(empty($all_sub)) return;
+
+        if($type=='public'){
+            $temp=[];
+            foreach ($all_sub as $v){
+                if(!is_array($v)) $temp[]=$v;
+            }
+            $global->save('add_sub',$this->resub($temp));
+        }else{
+
+        }
+    }
 }
diff --git a/src/Api/WebSocket/SocketServer.php b/src/Api/WebSocket/SocketServer.php
@@ -44,18 +44,22 @@ private function addConnection(string $tag,array $keysecret=[]){
 
     private function newConnection(){
         return function($tag,$keysecret){
+            $baseurl='ws://real.okex.com:8443/ws/v3';
+
             $global=$this->client();
 
-            $this->connection[$this->connectionIndex] = new AsyncTcpConnection('ws://real.okex.com:8443/ws/v3');
+            $this->connection[$this->connectionIndex] = new AsyncTcpConnection($baseurl);
             $this->connection[$this->connectionIndex]->transport = 'ssl';
 
+            $this->log('Connection '.$baseurl);
+
             //自定义属性
             $this->connection[$this->connectionIndex]->tag=$tag;//标记公共连接还是私有连接
             if(!empty($keysecret)) $this->connection[$this->connectionIndex]->tag_keysecret=$keysecret;//标记私有连接
 
             $this->connection[$this->connectionIndex]->onConnect=$this->onConnect($keysecret);
             $this->connection[$this->connectionIndex]->onMessage=$this->onMessage($global);
-            $this->connection[$this->connectionIndex]->onClose=$this->onClose();
+            $this->connection[$this->connectionIndex]->onClose=$this->onClose($global);
             $this->connection[$this->connectionIndex]->onError=$this->onError();
 
             $this->connect($this->connection[$this->connectionIndex]);
@@ -134,12 +138,20 @@ private function onMessage($global){
         };
     }
 
-    private function onClose(){
-        return function($con){
-            $this->log('reconnection');
+    private function onClose($global){
+        return function($con) use($global){
+            if($con->tag=='public'){
+                $this->log($con->tag.' reconnection');
+
+                $this->reconnection($global,'public');
+
+                //这里连接失败 会轮询 connect
+                $con->reConnect(10);
+            }else{
+                $this->log('connection close '.$con->tag_keysecret['key']);
 
-            //这里连接失败 会轮询 connect
-            $con->reConnect(5);
+                Timer::del($con->timer_other);
+            }
         };
     }
 
@@ -166,15 +178,40 @@ private function ping($con){
     private function other($con,$global){
         $time=isset($this->config['listen_time']) ? $this->config['listen_time'] : 2 ;
 
-        Timer::add($time, function() use($con,$global) {
+        $con->timer_other=Timer::add($time, function() use($con,$global) {
             $this->subscribe($con,$global);
 
             $this->unsubscribe($con,$global);
 
+            $this->debug($con,$global);
+
             $this->log('listen '.$con->tag);
         });
     }
 
+    /**
+     * 调试用
+     * @param $con
+     * @param $global
+     */
+    private function debug($con,$global){
+        if($con->tag=='public'){
+            //public
+            $debug=$global->get('debug');
+
+            if(isset($debug['public']) && $debug['public'][$con->tag]=='close'){
+                $this->log($con->tag.' debug '.json_encode($debug));
+
+                $debug['public'][$con->tag]='recon';
+                $global->save('debug',$debug);
+
+                $con->close();
+            }
+        }else{
+            //private
+        }
+    }
+
     private function subscribe($con,$global){
         if(empty($global->get('add_sub'))) return;
 
@@ -237,6 +274,7 @@ private function subscribePrivate($con,$global){
         $temp=$global->get('add_sub');
         //判断是否是私有连接，并判断该私有连接是否是  当前用户。
         foreach ($temp as $v){
+            if(!isset($v[1])) continue;
             $key=$v[1]['key'];
             if(count($v)>1 && $key==$keysecret['key']) $sub[]=$v[0];
         }

diff --git a/tests/websocket/client.php b/tests/websocket/client.php
@@ -48,7 +48,6 @@
             'spot/depth5:BCH-USDT',
             'futures/depth5:BCH-USD-210326',
             'swap/depth5:BCH-USD-SWAP',
-            'option/depth5:BTCUSD-20201021-11750-C',
         ]);
 
         break;
@@ -250,10 +249,7 @@
     }
 
     case 10005:{
-        $okex->keysecret($key_secret[0]);
-        $data=$okex->getSubscribes();
-        print_r(json_encode($data));
-        die;
+        $okex->client()->test_reconnection();
         break;
     }