STREAM_CLIENT_CONNECT | STREAM_CLIENT_PERSISTENT

applications/ChatDemo/Bootstrap/BusinessWorker.php

@@ -14,12 +14,6 @@ require_once ROOT_DIR . '/Lib/APLog.php';
 class BusinessWorker extends Man\Core\SocketWorker
 {
     /**
-     * BusinessWorker 实例
-     * @var BusinessWorker
-     */
-    protected static $instance = null;
-    
-    /**
      * 与gateway的连接
      * ['ip:port' => conn, 'ip:port' => conn, ...]
      * @var array
@@ -35,24 +29,6 @@ class BusinessWorker extends Man\Core\SocketWorker
         // 定时检查与gateway进程的连接
         \Man\Core\Lib\Task::init($this->event);
         \Man\Core\Lib\Task::add(1, array($this, 'checkGatewayConnections'));
-        self::$instance = $this;
-    }
-    
-    /**
-     * 获取BusinessWorker实例
-     * @return BusinessWorker
-     */
-    public static function instance()
-    {
-        return self::$instance;;
-    }
-    
-    /**
-     * 获取与gateway的连接
-     */
-    public static function getGatewayConnections()
-    {
-        return self::$gatewayConnections;
     }
     
     /**
@@ -132,26 +108,17 @@ class BusinessWorker extends Man\Core\SocketWorker
      */
     protected function closeClient($fd)
     {
-        foreach(self::$gatewayConnections as $con)
+        foreach(self::$gatewayConnections as $addr => $con)
         {
             $the_fd = (int) $con;
             if($the_fd == $fd)
             {
-                unset(self::$gatewayConnections[$fd]);
+                unset(self::$gatewayConnections[$addr]);
             }
         }
         parent::closeClient($fd);
     }
     
-    /**
-     * 向客户端发送数据
-     * @see Man\Core.SocketWorker::sendToClient()
-     */
-    public function sendToClient($buffer, $fd)
-    {
-        $this->currentDealFd = (int)$fd;
-        parent::sendToClient($buffer);
-    }
 }
 
 

applications/ChatDemo/Lib/Gateway.php

@@ -28,9 +28,10 @@ class GateWay
        $pack->header['uid'] = Context::$uid;
        $pack->body = (string)$message;
        $buffer = $pack->getBuffer();
-       foreach(BusinessWorker::getGatewayConnections() as $con)
+       $all_addresses = Store::get('GLOBAL_GATEWAY_ADDRESS');
+       foreach($all_addresses as $address)
        {
-           BusinessWorker::instance()->sendToClient($buffer, $con);
+           self::sendToGateway($address, $buffer);
        }
    }
    
@@ -197,13 +198,8 @@ class GateWay
     */
    public static function sendToGateway($address, $buffer)
    {
-       $connections = BusinessWorker::instance()->getGatewayConnections();
-       if(!isset($connections[$address]))
-       {
-           $e = new \Exception("sendToGateway($address, $buffer) fail; getGatewayConnections:".json_encode($connections));
-           APLog::add($e->__toString());
-           return false;
-       }
-       return BusinessWorker::instance()->sendToClient($buffer, $connections[$address]);
+       $client = stream_socket_client($address, $errno, $errmsg, 1, STREAM_CLIENT_CONNECT | STREAM_CLIENT_PERSISTENT);
+       $len = stream_socket_sendto($client, $buffer);
+       return $len == strlen($buffer);
    }
 }