fix redis disconnect

helper/queue/iqueue.php

@@ -27,35 +27,29 @@ class IQueueDB
         $this->_tb_prefix = $queue_name;
         $this->_comode = $comode;
 
-        $this->_redis = null;
+        if ($this->_comode) {
+            $this->_redis = new \Swoole\Coroutine\Redis();
+        } else {
+            $this->_redis = new Redis();
+        }
+
         $this->connect();
     }
 
     public function connect() : bool
     {
-        if($this->_redis != null && $this->_redis->connected) return true;
+        if($this->_redis->connected) return true;
 
-        if($this->_comode) {
-            $this->_redis  = new \Swoole\Coroutine\Redis();
+        if ($this->_comode) {
             $ret = $this->_redis->connect(C('coroutine.redis_host'), C('coroutine.redis_port'));
-            Log::record("Swoole\Coroutine\Redis connect ret = {$ret}",Log::DEBUG);
-        }
-        else {
-            $this->_redis = new Redis();
+        } else {
             $ret = $this->_redis->connect(C('queue.host'), C('queue.port'), 20);
-            $this->_redis->setOption(Redis::OPT_READ_TIMEOUT, 10);
         }
+        Log::record("Swoole\Coroutine\Redis connect ret = {$ret}",Log::DEBUG);
 
         return $this->_redis->connected;
     }
 
-    public function close()
-    {
-        if($this->_redis != null) {
-            $this->_redis->close();
-        }
-    }
-
     public function rpush($value)
     {
         try {
@@ -122,6 +116,7 @@ class IClient
     {
         $this->mQueuedb = $queueDb;
     }
+
     public function push($key, $value)
     {
         return $this->mQueuedb->lpush(serialize([$key=>$value]));
@@ -140,10 +135,6 @@ class IServer
     {
         return $this->_queuedb->connect();
     }
-    public function close()
-    {
-        $this->_queuedb->close();
-    }
 
     public function pop($key,$time)
     {

helper/refill/policy/IPolicy.php

@@ -10,6 +10,6 @@ interface IPolicy
     public function can_retry(array $refill_info, array $order_info) : array;
     public function first_quality($mchid): int;
     public function allow($mchid,$card_type,$amount,$quality) : bool;
-    public function can_nettry($quality,$order_time,$commit_times) : bool;
+    public function can_nettry($mchid,$quality,$order_time,$commit_times) : bool;
     public function notify($order_info, $refill_info) : bool;
 }

helper/refill/policy/lingzh/policy.php

@@ -14,17 +14,19 @@ class policy extends ProviderManager implements IPolicy
     const high_time_out = 30;
 
     protected $mChannelControl;
-
+    protected $mMchctl;
     public function __construct()
     {
         parent::__construct();
         $this->mChannelControl = new chctl();
+        $this->mMchctl = new mchctl();
     }
 
     public function load()
     {
         parent::load();
         $this->mChannelControl->load();
+        $this->mMchctl->load();
     }
 
     public function find_providers(int $spec, int $card_type,int $quality): array
@@ -57,8 +59,9 @@ class policy extends ProviderManager implements IPolicy
 
     public function first_quality($mchid): int
     {
-
+        return 1;
     }
+
     public function can_retry(array $refill_info, array $order_info): array
     {
         return [false,0];
@@ -69,7 +72,7 @@ class policy extends ProviderManager implements IPolicy
         return true;
     }
 
-    public function can_nettry($quality,$order_time,$commit_times) : bool
+    public function can_nettry($mchid,$quality,$order_time,$commit_times) : bool
     {
         if($quality == 1) {
             return $commit_times < policy::normal_times && time() - $order_time < policy::normle_time_out;

helper/refill/policy/xyz/policy.php

@@ -123,6 +123,27 @@ class policy extends ProviderManager implements IPolicy
         }
     }
 
+    public function can_nettry($mchid,$quality,$order_time,$commit_times) : bool
+    {
+        [$sucdess,$quality,$time_out] = $this->mMchctl->getCtls($mchid);
+
+        if($sucdess)
+        {
+            if($quality == 1) {
+                return $commit_times < policy::normal_times && time() - $order_time < $time_out;
+            }
+            else {
+                return $commit_times < policy::high_times && time() - $order_time < $time_out;
+            }
+        }
+        elseif($quality == 1) {
+            return $commit_times < policy::normal_times && time() - $order_time < policy::normle_time_out;
+        }
+        else {
+            return $commit_times < policy::high_times && time() - $order_time < policy::high_time_out;
+        }
+    }
+
     public function first_quality($mchid): int
     {
         [$sucdess,$quality,$time_out] = $this->mMchctl->getCtls($mchid);
@@ -133,7 +154,6 @@ class policy extends ProviderManager implements IPolicy
         }
     }
 
-
     public function allow($mchid,$card_type,$amount,$quality) : bool
     {
         $reader = function () {
@@ -168,15 +188,7 @@ class policy extends ProviderManager implements IPolicy
             return true;
         }
     }
-    public function can_nettry($quality,$order_time,$commit_times) : bool
-    {
-        if($quality == 1) {
-            return $commit_times < policy::normal_times && time() - $order_time < policy::normle_time_out;
-        }
-        else {
-            return $commit_times < policy::high_times && time() - $order_time < policy::high_time_out;
-        }
-    }
+
     public function notify($order_info, $refill_info) : bool
     {
         $order_state = $order_info['order_state'];

mobile/control/refill.php

@@ -68,8 +68,6 @@ class refillControl extends merchantControl
         $notify_url = $_GET['notifyurl'];
         $mch_order = $_GET['order_sn']; //对方的order编号
 
-        $quality = refill\RefillFactory::instance()->first_quality($this->mchid());
-
         if(!$this->check_mchorder($this->mchid(),$mch_order)) {
             return self::outerr(205,"客户订单号重复或者为空.");
         }
@@ -98,8 +96,7 @@ class refillControl extends merchantControl
             'amount' => $amount,
             'card_no' => $card_no,
             'mch_order' => $mch_order,
-            'notify_url' => $notify_url,
-            'quality' => $quality];
+            'notify_url' => $notify_url];
 
         $state = refill\util::push_add($params);
         if($state === true) {

racc/control/lzrefill.php

@@ -74,7 +74,6 @@ class lzrefillControl extends lzbaseControl
         $card_no = $_GET['mob'];
         $notify_url = $_GET['notifyurl'];
         $mch_order = $_GET['ord']; //对方的order编号
-        $quality = $_GET['quality'];
 
         //此处判断是对的
         if($amount > $this->member_available) {
@@ -94,8 +93,7 @@ class lzrefillControl extends lzbaseControl
             'amount' => $amount,
             'card_no' => $card_no,
             'mch_order' => $mch_order,
-            'notify_url' => $notify_url,
-            'quality' => $quality];
+            'notify_url' => $notify_url];
 
         $ret = refill\util::push_add($params);
         if($ret) {

rdispatcher/codispatcher.php

@@ -44,11 +44,19 @@ function handle_error($level, $message, $file, $line)
 
 function sub_message($channels)
 {
+    $redis = new Swoole\Coroutine\Redis();
+
     while (true)
     {
-        try {
-            $redis = new Swoole\Coroutine\Redis();
-            $ret = $redis->connect(C('coroutine.redis_host'), C('coroutine.redis_port'));
+        try
+        {
+            if(!$redis->connected) {
+                $ret = $redis->connect(C('coroutine.redis_host'), C('coroutine.redis_port'));
+                Log::record("sub_message redis connected = {$redis->connected}",Log::DEBUG);
+            }
+            else {
+                $ret = true;
+            }
 
             if(!$ret) {
                 Log::record("sub_message cannot connet redis.",Log::DEBUG);
@@ -57,28 +65,22 @@ function sub_message($channels)
             }
             elseif($redis->subscribe($channels))
             {
-                Log::record("sub_message starting recv",Log::DEBUG);
                 while ($msg = $redis->recv())
                 {
                     [$sub_type, $channel, $content] = $msg;
-                    Log::record("sub_message recv mgs:{$sub_type}-{$channel}-{$content}",Log::DEBUG);
-
                     $content = unserialize($content);
                     $type = $content['type'];
-                    Log::record("sub_message recv mgs:{$sub_type}-{$channel}-{$type}",Log::DEBUG);
 
                     if($channel != 'refill') continue;
-
                     if($type == 'channel' || $type == 'merchant') {
-                        Log::record("recv publish message reload refill",Log::DEBUG);
+                        Log::record("sub_message recv mgs:{$sub_type}-{$channel}-{$type}",Log::DEBUG);
                         refill\RefillFactory::instance()->load();
                     }
                 }
-                $redis->close();
-                Log::record("sub_message redis close",Log::DEBUG);
+                Log::record("sub_message redis recv timeout",Log::DEBUG);
             }
             else {
-                Log::record("subscribe publish message error",Log::ERR);
+                Log::record("sub_message subscribe error",Log::ERR);
             }
         }
         catch (Exception $ex)
@@ -88,38 +90,37 @@ function sub_message($channels)
     }
 }
 
+//Co\run(function()
+//{
+//    Base::run_util();
+//    set_error_handler('handle_error');
+//    go(function () {
+//        sub_message(['refill']);
+//    });
+//    $looper = new processor(true);
+//    $looper->run();
+//});
+
+for ($i = 0; $i < $process_count;$i++)
+{
+    $process = new Swoole\Process(function(Swoole\Process $worker)
+    {
+        Base::run_util();
+        set_error_handler('handle_error');
+        go(function () {
+            sub_message(['refill']);
+        });
+
+        $looper = new processor(true);
+        $looper->run();
+
+    }, false, false, true);
 
-Co\run(function()
+    $process->start();
+}
+
+for ($i = 0; $i < $process_count;$i++)
 {
-    Base::run_util();
-    set_error_handler('handle_error');
-    go(function () {
-        sub_message(['refill']);
-    });
-    $looper = new processor(true);
-    $looper->run();
-});
-
-//for ($i = 0; $i < $process_count;$i++)
-//{
-//    $process = new Swoole\Process(function(Swoole\Process $worker)
-//    {
-//        Base::run_util();
-//        set_error_handler('handle_error');
-//        go(function () {
-//            sub_message(['refill']);
-//        });
-//
-////        $looper = new processor(true);
-////        $looper->run();
-//
-//    }, false, false, true);
-//
-//    $process->start();
-//}
-
-//for ($i = 0; $i < $process_count;$i++)
-//{
-//    $status = Swoole\Process::wait(true);
-//    Log::record("Recycled #{$status['pid']}, code={$status['code']}, signal={$status['signal']}",Log::DEBUG);
-//}
+    $status = Swoole\Process::wait(true);
+    Log::record("Recycled #{$status['pid']}, code={$status['code']}, signal={$status['signal']}",Log::DEBUG);
+}

rdispatcher/proxy.php

@@ -17,7 +17,6 @@ class proxy
         $amount = intval($params['amount']);
 
         $card_no = $params['card_no'];
-        $quality = intval($params['quality'] ?? 1);
         $mch_order = $params['mch_order'];
         $notify_url = $params['notify_url'];
         $idcard = $params['idcard'] ?? '';
@@ -27,6 +26,13 @@ class proxy
         $commit_times = $params['commit_times'] ?? 0;
         $last_order_id = $params['order_id'] ?? 0;
 
+        $quality = intval($params['quality']);
+
+        if($quality != 1 && $quality != 2) {
+            $quality =  refill\RefillFactory::instance()->first_quality($mchid);
+        }
+
+
         [$errcode, $errmsg,$order_id,$neterr] = refill\RefillFactory::instance()->add($mchid, $buyer_id, $amount, $card_no,
             $mch_order, $idcard, $card_name, $notify_url,$quality, $order_time, $commit_times,$last_order_id);