onMessage 事件回调
@access public
@param  \Workerman\Connection\TcpConnection    $connection
@param  mixed                                  $data
@return void
760656
public function onMessage($connection, $data)
    {
        $uri  = parse_url($_SERVER['REQUEST_URI']);
        $path = isset($uri['path']) ? $uri['path'] : '/';

        $file = $this->root . $path;

        if (!is_file($file)) {
            $this->app->worker($connection, $data);
        } else {
            $this->sendFile($connection, $file);
        }
    }
733602
protected function fireMessageEvent($handler, $connection, $frame)
    {
        $isBinary = $frame->opcode == WEBSOCKET_OPCODE_BINARY;

        $this->fireWebSocketHandler($handler, 'onMessage', $connection, $isBinary, $frame->data);
    }
811867
public function handleStop(ConnectionInterface $conn, array $data)
    {
        $connSubscriptions = $this->connStorage->offsetGet($conn);
        $subscription = array_get($connSubscriptions, $data['id']);

        if (!is_null($subscription)) {
            unset($this->subscriptions[$subscription['name']][$subscription['index']]);
            unset($connSubscriptions[$subscription['id']]);
            $this->connStorage->offsetSet($conn, $connSubscriptions);
            $this->callListener(ON_DISCONNECT, [$subscription, $this->rootValue, $this->context]);
        }
    }
91430
public function handleData($data, $conn) {
        try {
            $this->app->onMessage($conn->decor, $data);
        } catch (\Exception $e) {
            $this->handleError($e, $conn);
        }
    }
614531
public function onMessage($connection, $data)
    {
        Base::getLog()->debug(__METHOD__ . ' receive http request', [
            'uri'  => $_SERVER['REQUEST_URI'],
            'ip'   => $connection->getRemoteIp(),
            'port' => $connection->getRemotePort(),
            'data' => $data,
        ]);

        // 请求的文件
        $urlInfo = parse_url($_SERVER['REQUEST_URI']);
        if ( ! $urlInfo)
        {
            Base::getHttp()->header('HTTP/1.1 400 Bad Request');
            Base::getLog()->warning(__METHOD__ . ' receive bad request', [
                'uri'  => $_SERVER['REQUEST_URI'],
                'ip'   => $connection->getRemoteIp(),
                'port' => $connection->getRemotePort(),
            ]);
            return $connection->close($this->error400);
        }

        $path = $urlInfo['path'];

        $pathInfo = pathinfo($path);
        $extension = isset($pathInfo['extension']) ? $pathInfo['extension'] : '';
        if ($extension === '')
        {
            $path = ($len = strlen($path)) && $path[$len - 1] === '/'
                ? $path . $this->indexFile
                : $path . '/' . $this->indexFile;
            $extension = 'php';
        }

        $serverName = Arr::get($_SERVER, 'SERVER_NAME');

        $rootDir = isset($this->serverRoot[$serverName])
            ? $this->serverRoot[$serverName]
            : current($this->serverRoot);

        $file = "$rootDir/$path";

        // 对应的php文件不存在，而且支持rewrite
        if ( ! is_file($file) && $this->rewrite)
        {
            $file = is_string($this->rewrite)
                ? $rootDir . '/' . $this->rewrite
                : $rootDir . '/' . $this->indexFile;
            $extension = 'php';
            $_SERVER['PATH_INFO'] = $_SERVER['REQUEST_URI'];
        }

        // 请求的文件存在
        if (is_file($file))
        {
            Base::getLog()->debug(__METHOD__ . ' request file existed', [
                'file'      => $file,
                'extension' => $extension,
            ]);

            // 判断是否是站点目录里的文件
            if (( ! ($requestRealPath = realpath($file)) || ! ($rootDirRealPath = realpath($rootDir))) || 0 !== strpos($requestRealPath, $rootDirRealPath))
            {
                Base::getHttp()->header('HTTP/1.1 400 Bad Request');
                Base::getLog()->warning(__METHOD__ . ' receive bad request', [
                    'uri'  => $_SERVER['REQUEST_URI'],
                    'ip'   => $connection->getRemoteIp(),
                    'port' => $connection->getRemotePort(),
                ]);
                return $connection->close('<h1>400 Bad Request</h1>');
            }

            $file = realpath($file);

            // 如果请求的是php文件
            // PHP文件需要include
            if ($extension === 'php')
            {
                Base::getLog()->debug(__METHOD__ . ' handle request', [
                    'uri'  => $_SERVER['REQUEST_URI'],
                    'ip'   => $connection->getRemoteIp(),
                    'port' => $connection->getRemotePort(),
                    'file' => $file,
                ]);

                Base::getLog()->debug(__METHOD__ . ' clean components - start');
                Base::cleanComponents();
                Base::getLog()->debug(__METHOD__ . ' clean components - end');

                $cwd = getcwd();
                chdir($rootDir);
                ini_set('display_errors', 'off');
                // 缓冲输出
                ob_start();
                // 载入php文件
                try
                {
                    // $_SERVER变量
                    $_SERVER['HOME'] = $_SERVER['DOCUMENT_ROOT'] = dirname($file);
                    $_SERVER['SCRIPT_FILENAME'] = $file;
                    Base::getLog()->debug(__METHOD__ . ' dispatch client info', [
                        'ip'   => $_SERVER['REMOTE_ADDR'],
                        'port' => $_SERVER['REMOTE_PORT'],
                    ]);

                    include $file;
                }
                catch (Exception $e)
                {
                    Base::getLog()->error($e->getMessage(), [
                        'code' => $e->getCode(),
                        'file' => $e->getFile(),
                        'line' => $e->getLine(),
                    ]);
                    // 如果不是exit
                    if ($e->getMessage() != 'jump_exit')
                    {
                        echo $e;
                    }
                }

                Patch::applyShutdownFunction();

                $content = ob_get_clean();
                ini_set('display_errors', 'on');
                $result = $connection->close($content);
                chdir($cwd);
                return $result;
            }
            // 请求的是静态资源文件
            else
            {
                $contentType = Mime::getMimeFromExtension($extension, self::$defaultMimeType);
                Base::getLog()->debug(__METHOD__ . ' get static file content type', [
                    'extension'   => $extension,
                    'contentType' => $contentType,
                ]);
                Base::getHttp()->header('Content-Type: ' . $contentType);

                // 获取文件信息
                $info = stat($file);

                $modifiedTime = $info ? date('D, d M Y H:i:s', Arr::get($info, 'mtime')) . ' GMT' : '';

                // 如果有$_SERVER['HTTP_IF_MODIFIED_SINCE']
                if ( ! empty($_SERVER['HTTP_IF_MODIFIED_SINCE']) && $info)
                {
                    // 文件没有更改则直接304
                    if ($modifiedTime === $_SERVER['HTTP_IF_MODIFIED_SINCE'])
                    {
                        Base::getLog()->debug(__METHOD__ . ' no modified, return 304');
                        // 304
                        Base::getHttp()->header('HTTP/1.1 304 Not Modified');
                        // 发送给客户端
                        return $connection->close('');
                    }
                }

                if ($modifiedTime)
                {
                    Base::getLog()->debug(__METHOD__ . ' set last modified time', [
                        'time' => $modifiedTime,
                    ]);
                    Base::getHttp()->header("Last-Modified: $modifiedTime");
                }
                // 发送给客户端
                return $connection->close(file_get_contents($file));
            }
        }
        else
        {
            Base::getLog()->warning(__METHOD__ . ' requested file not found', [
                'file' => $file,
            ]);

            // 404
            Base::getHttp()->header("HTTP/1.1 404 Not Found");
            return $connection->close($this->error404);
        }
    }
40916
protected function checkSlave(Connection $connection)
    {
        $paused = 0;

        while (($lag = $this->getLag($connection)) > $this->options['max_lag']) {
            $this->paused = true;
            $for = $this->options['pause_lag'];

            $this->logger->notice('Chunk detected lag of {lag} on {database}, pausing for {for}', [
                'lag'      => $lag,
                'database' => sprintf('%s/%s', $connection->getHost(), $connection->getDatabase()),
                'for'      => $for
            ]);

            usleep($for);
            $paused += $for;

            if ($paused > $this->options['max_pause_lag']) {
                if ($this->options['continue_lag']) {
                    break;
                } else {
                    throw new RuntimeException('Slave lag did not recover after processing chunk. Aborting.');
                }
            }
        }
    }
132089
protected function triggerEvent($event, $source = null, $data = null) {
		$this->getEventsManager()->trigger($event, $source, $data);
	}
1251414
function onMessage(ProcConnection $conn, $data) {
        if ($this->_status == self::STATUS_INIT)
            return;
        $data = self::_parseTex($data);
        if ($data === false)
            return;
        if ($this->_status == self::STATUS_FORMULA) {
            $this->_formula = $data;
            $this->_id = self::genPic($data);
            $this->_status = self::STATUS_RESULT;
            $conn->send($this->_current->withEval());
        } else {
            $message = $this->_current->getMessage();
            msg_send(Captcha::getQueue(), 1, [$this->_id, $message, $data]);
            if (msg_stat_queue(Captcha::getQueue())['msg_qnum'] < Config::get('CAPTCHA_CACHE_MAX')) {
                $this->generate();
            } else {
                $this->_status = self::STATUS_INIT;
            }
        }
    }
678231
public function onMessage(ConnectionInterface $from, $message)
    {
        $incoming_client = $this->clients->findByConnection($from);

        if ($incoming_client) {
            // Handle session.
            // $incoming_client->session_manager->handle();

            // Handle message.
            $message = new Message($message);

            // Fire the event.
            Event::fire(new MessageReceived($this->server, $this->clients, $incoming_client, $message));
        }
    }
483638
public function _eventHandlerPeek($event, $source, $data)
    {
        $context = $this->_context;

        $context->events[] = $event;

        if ($event === 'logger:log') {
            /**
             * @var Log $log
             */
            $log = $data;
            $format = '[%time%][%level%] %message%';
            $replaces = [
                '%time%' => date('H:i:s.', $log->timestamp) . sprintf('%.03d', ($log->timestamp - (int)$log->timestamp) * 1000),
                '%level%' => $log->level,
                '%message%' => $log->message
            ];
            $context->log[] = [
                'level' => $log->level,
                'message' => strtr($format, $replaces)
            ];
        } elseif ($event === 'db:beforeQuery' || $event === 'db:beforeExecute') {
            /**
             * @var \ManaPHP\DbInterface $source
             */
            $preparedSQL = $source->getSQL();
            if (!isset($context->sql_prepared[$preparedSQL])) {
                $context->sql_prepared[$preparedSQL] = 1;
            } else {
                $context->sql_prepared[$preparedSQL]++;
            }

            $context->sql_count++;
            $context->sql_executed[] = [
                'prepared' => $source->getSQL(),
                'bind' => $source->getBind(),
                'emulated' => $source->getEmulatedSQL()
            ];
        } elseif ($event === 'db:afterQuery' || $event === 'db:afterExecute') {
            /**
             * @var \ManaPHP\DbInterface $source
             */
            $context->sql_executed[$context->sql_count - 1]['elapsed'] = $data['elapsed'];
            $context->sql_executed[$context->sql_count - 1]['row_count'] = $source->affectedRows();
        } elseif ($event === 'db:beginTransaction' || $event === 'db:rollbackTransaction' || $event === 'db:commitTransaction') {
            $context->sql_count++;

            $parts = explode(':', $event);
            $name = $parts[1];

            $context->sql_executed[] = [
                'prepared' => $name,
                'bind' => [],
                'emulated' => $name,
                'time' => 0,
                'row_count' => 0
            ];

            $preparedSQL = $name;
            if (!isset($context->_sql_prepared[$preparedSQL])) {
                $context->sql_prepared[$preparedSQL] = 1;
            } else {
                $context->sql_prepared[$preparedSQL]++;
            }
        } elseif ($event === 'renderer:beforeRender') {
            $vars = $data['vars'];
            foreach ((array)$vars as $k => $v) {
                if ($v instanceof Component) {
                    unset($vars[$k]);
                }
            }
            unset($vars['di']);
            $context->view[] = ['file' => $data['file'], 'vars' => $vars, 'base_name' => basename(dirname($data['file'])) . '/' . basename($data['file'])];
        } elseif ($event === 'mongodb:afterQuery') {
            $item = [];
            $item['type'] = 'query';
            $item['raw'] = ['namespace' => $data['namespace'], 'filter' => $data['filter'], 'options' => $data['options']];
            $options = $data['options'];
            list(, $collection) = explode('.', $data['namespace'], 2);
            $shell = "db.$collection.";
            $shell .= (isset($options['limit']) ? 'findOne(' : 'find(') . json_encode($data['filter'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
            if (isset($options['projection'])) {
                $shell .= ', ' . json_encode($options['projection'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) . ');';
            } else {
                $shell .= ');';
            }

            $item['shell'] = $shell;
            $item['elapsed'] = $data['elapsed'];
            $context->mongodb[] = $item;
        } elseif ($event === 'mongodb:afterCommand') {
            $item = [];
            $item['type'] = 'command';
            $item['raw'] = ['db' => $data['db'], 'command' => $data['command'], 'options' => $data['options']];
            $item['shell'] = [];
            $item['elapsed'] = $data['elapsed'];
            $context->mongodb[] = $item;
        } elseif ($event === 'mongodb:afterBulkWrite') {
            $item = [];
            $item['type'] = 'bulkWrite';
            $item['raw'] = ['db' => $data['db'], 'command' => $data['command'], 'options' => $data['options']];
            $item['shell'] = [];
            $item['elapsed'] = $data['elapsed'];
            $context->mongodb = [];
        }
    }
1367539
public function onConnectGateway($connection)
    {
        $this->gatewayConnections[$connection->remoteAddress] = $connection;
        unset($this->_connectingGatewayAddresses[$connection->remoteAddress], $this->_waitingConnectGatewayAddresses[$connection->remoteAddress]);
    }
848203
public function setConnection($name, ConnectionInterface $connection)
    {
        if (null !== $this->loggerCallable) {
            $connection->setLoggerCallable($this->loggerCallable);
            $connection->setLogDefault(array('connection' => $name));
        } else {
            $connection->setLoggerCallable(null);
        }

        $this->connections[$name] = $connection;
    }
964125
protected function initializeConnection($remote, $connection)
    {
        try {
            $context = $this->getContext($connection);
            $socket = $this->getSocket($remote, $context);
            $stream = $this->getStream($socket);
            $this->initializeStream($stream, $connection);
            $this->addActiveConnection($connection);
        } catch (\Exception $e) {
            $this->emitConnectionError($e, $connection);
        }

        $this->emit('connect.after.each', array($connection, $connection->getData('write')));
    }
268727
public function sendAsyncByDb(\PDO $connection) : void
    {
        if ($this->data) {
            throw new DataNotSupportedException();
        }

        $connection->exec('NOTIFY ' . static::NAME);
    }
653789
public function onWorkerMessage($connection, $data)
    {
        $cmd = $data['cmd'];
        if (empty($connection->authorized) && $cmd !== GatewayProtocol::CMD_WORKER_CONNECT && $cmd !== GatewayProtocol::CMD_GATEWAY_CLIENT_CONNECT) {
            self::log("Unauthorized request from " . $connection->getRemoteIp() . ":" . $connection->getRemotePort());
            return $connection->close();
        }
        switch ($cmd) {
            // BusinessWorker连接Gateway
            case GatewayProtocol::CMD_WORKER_CONNECT:
                $worker_info = json_decode($data['body'], true);
                if ($worker_info['secret_key'] !== $this->secretKey) {
                    self::log("Gateway: Worker key does not match ".var_export($this->secretKey, true)." !== ". var_export($this->secretKey));
                    return $connection->close();
                }
                $key = $connection->getRemoteIp() . ':' . $worker_info['worker_key'];
                // 在一台服务器上businessWorker->name不能相同
                if (isset($this->_workerConnections[$key])) {
                    self::log("Gateway: Worker->name conflict. Key:{$key}");
		    $connection->close();
                    return;
                }
		$connection->key = $key;
                $this->_workerConnections[$key] = $connection;
                $connection->authorized = true;
                return;
            // GatewayClient连接Gateway
            case GatewayProtocol::CMD_GATEWAY_CLIENT_CONNECT:
                $worker_info = json_decode($data['body'], true);
                if ($worker_info['secret_key'] !== $this->secretKey) {
                    self::log("Gateway: GatewayClient key does not match ".var_export($this->secretKey, true)." !== ".var_export($this->secretKey, true));
                    return $connection->close();
                }
                $connection->authorized = true;
                return;
            // 向某客户端发送数据，Gateway::sendToClient($client_id, $message);
            case GatewayProtocol::CMD_SEND_TO_ONE:
                if (isset($this->_clientConnections[$data['connection_id']])) {
                    $this->_clientConnections[$data['connection_id']]->send($data['body']);
                }
                return;
            // 踢出用户，Gateway::closeClient($client_id, $message);
            case GatewayProtocol::CMD_KICK:
                if (isset($this->_clientConnections[$data['connection_id']])) {
                    $this->_clientConnections[$data['connection_id']]->close($data['body']);
                }
                return;
            // 立即销毁用户连接, Gateway::destroyClient($client_id);
            case GatewayProtocol::CMD_DESTROY:
                if (isset($this->_clientConnections[$data['connection_id']])) {
                    $this->_clientConnections[$data['connection_id']]->destroy();
                }
                return;
            // 广播, Gateway::sendToAll($message, $client_id_array)
            case GatewayProtocol::CMD_SEND_TO_ALL:
                $raw = (bool)($data['flag'] & GatewayProtocol::FLAG_NOT_CALL_ENCODE);
                $body = $data['body'];
                if (!$raw && $this->protocolAccelerate && $this->protocol) {
                    $body = $this->preEncodeForClient($body);
                    $raw = true;
                }
                $ext_data = $data['ext_data'] ? json_decode($data['ext_data'], true) : '';
                // $client_id_array 不为空时，只广播给 $client_id_array 指定的客户端
                if (isset($ext_data['connections'])) {
                    foreach ($ext_data['connections'] as $connection_id) {
                        if (isset($this->_clientConnections[$connection_id])) {
                            $this->_clientConnections[$connection_id]->send($body, $raw);
                        }
                    }
                } // $client_id_array 为空时，广播给所有在线客户端
                else {
                    $exclude_connection_id = !empty($ext_data['exclude']) ? $ext_data['exclude'] : null;
                    foreach ($this->_clientConnections as $client_connection) {
                        if (!isset($exclude_connection_id[$client_connection->id])) {
                            $client_connection->send($body, $raw);
                        }
                    }
                }
                return;
            // 重新赋值 session
            case GatewayProtocol::CMD_SET_SESSION:
                if (isset($this->_clientConnections[$data['connection_id']])) {
                    $this->_clientConnections[$data['connection_id']]->session = $data['ext_data'];
                }
                return;
            // session合并
            case GatewayProtocol::CMD_UPDATE_SESSION:
                if (!isset($this->_clientConnections[$data['connection_id']])) {
                    return;
                } else {
                    if (!$this->_clientConnections[$data['connection_id']]->session) {
                        $this->_clientConnections[$data['connection_id']]->session = $data['ext_data'];
                        return;
                    }
                    $session = Context::sessionDecode($this->_clientConnections[$data['connection_id']]->session);
                    $session_for_merge = Context::sessionDecode($data['ext_data']);
                    $session = $session_for_merge + $session;
                    $this->_clientConnections[$data['connection_id']]->session = Context::sessionEncode($session);
                }
                return;
            case GatewayProtocol::CMD_GET_SESSION_BY_CLIENT_ID:
                if (!isset($this->_clientConnections[$data['connection_id']])) {
                    $session = serialize(null);
                } else {
                    if (!$this->_clientConnections[$data['connection_id']]->session) {
                        $session = serialize(array());
                    } else {
                        $session = $this->_clientConnections[$data['connection_id']]->session;
                    }
                }
                $connection->send(pack('N', strlen($session)) . $session, true);
                return;
            // 获得客户端在线状态 Gateway::getALLClientInfo()
            case GatewayProtocol::CMD_GET_ALL_CLIENT_INFO:
                $client_info_array = array();
                foreach ($this->_clientConnections as $connection_id => $client_connection) {
                    $client_info_array[$connection_id] = $client_connection->session;
                }
                $buffer = serialize($client_info_array);
                $connection->send(pack('N', strlen($buffer)) . $buffer, true);
                return;
            // 判断某个 client_id 是否在线 Gateway::isOnline($client_id)
            case GatewayProtocol::CMD_IS_ONLINE:
                $buffer = serialize((int)isset($this->_clientConnections[$data['connection_id']]));
                $connection->send(pack('N', strlen($buffer)) . $buffer, true);
                return;
            // 将 client_id 与 uid 绑定
            case GatewayProtocol::CMD_BIND_UID:
                $uid = $data['ext_data'];
                if (empty($uid)) {
                    echo "bindUid(client_id, uid) uid empty, uid=" . var_export($uid, true);
                    return;
                }
                $connection_id = $data['connection_id'];
                if (!isset($this->_clientConnections[$connection_id])) {
                    return;
                }
                $client_connection = $this->_clientConnections[$connection_id];
                if (isset($client_connection->uid)) {
                    $current_uid = $client_connection->uid;
                    unset($this->_uidConnections[$current_uid][$connection_id]);
                    if (empty($this->_uidConnections[$current_uid])) {
                        unset($this->_uidConnections[$current_uid]);
                    }
                }
                $client_connection->uid                      = $uid;
                $this->_uidConnections[$uid][$connection_id] = $client_connection;
                return;
            // client_id 与 uid 解绑 Gateway::unbindUid($client_id, $uid);
            case GatewayProtocol::CMD_UNBIND_UID:
                $connection_id = $data['connection_id'];
                if (!isset($this->_clientConnections[$connection_id])) {
                    return;
                }
                $client_connection = $this->_clientConnections[$connection_id];
                if (isset($client_connection->uid)) {
                    $current_uid = $client_connection->uid;
                    unset($this->_uidConnections[$current_uid][$connection_id]);
                    if (empty($this->_uidConnections[$current_uid])) {
                        unset($this->_uidConnections[$current_uid]);
                    }
                    $client_connection->uid_info = '';
                    $client_connection->uid      = null;
                }
                return;
            // 发送数据给 uid Gateway::sendToUid($uid, $msg);
            case GatewayProtocol::CMD_SEND_TO_UID:
                $uid_array = json_decode($data['ext_data'], true);
                foreach ($uid_array as $uid) {
                    if (!empty($this->_uidConnections[$uid])) {
                        foreach ($this->_uidConnections[$uid] as $connection) {
                            /** @var TcpConnection $connection */
                            $connection->send($data['body']);
                        }
                    }
                }
                return;
            // 将 $client_id 加入用户组 Gateway::joinGroup($client_id, $group);
            case GatewayProtocol::CMD_JOIN_GROUP:
                $group = $data['ext_data'];
                if (empty($group)) {
                    echo "join(group) group empty, group=" . var_export($group, true);
                    return;
                }
                $connection_id = $data['connection_id'];
                if (!isset($this->_clientConnections[$connection_id])) {
                    return;
                }
                $client_connection = $this->_clientConnections[$connection_id];
                if (!isset($client_connection->groups)) {
                    $client_connection->groups = array();
                }
                $client_connection->groups[$group]               = $group;
                $this->_groupConnections[$group][$connection_id] = $client_connection;
                return;
            // 将 $client_id 从某个用户组中移除 Gateway::leaveGroup($client_id, $group);
            case GatewayProtocol::CMD_LEAVE_GROUP:
                $group = $data['ext_data'];
                if (empty($group)) {
                    echo "leave(group) group empty, group=" . var_export($group, true);
                    return;
                }
                $connection_id = $data['connection_id'];
                if (!isset($this->_clientConnections[$connection_id])) {
                    return;
                }
                $client_connection = $this->_clientConnections[$connection_id];
                if (!isset($client_connection->groups[$group])) {
                    return;
                }
                unset($client_connection->groups[$group], $this->_groupConnections[$group][$connection_id]);
                return;
            // 向某个用户组发送消息 Gateway::sendToGroup($group, $msg);
            case GatewayProtocol::CMD_SEND_TO_GROUP:
                $raw = (bool)($data['flag'] & GatewayProtocol::FLAG_NOT_CALL_ENCODE);
                $body = $data['body'];
                if (!$raw && $this->protocolAccelerate && $this->protocol) {
                    $body = $this->preEncodeForClient($body);
                    $raw = true;
                }
                $ext_data = json_decode($data['ext_data'], true);
                $group_array = $ext_data['group'];
                $exclude_connection_id = $ext_data['exclude'];

                foreach ($group_array as $group) {
                    if (!empty($this->_groupConnections[$group])) {
                        foreach ($this->_groupConnections[$group] as $connection) {
                            if(!isset($exclude_connection_id[$connection->id]))
                            {
                                /** @var TcpConnection $connection */
                                $connection->send($body, $raw);
                            }
                        }
                    }
                }
                return;
            // 获取某用户组成员信息 Gateway::getClientInfoByGroup($group);
            case GatewayProtocol::CMD_GET_CLINET_INFO_BY_GROUP:
                $group = $data['ext_data'];
                if (!isset($this->_groupConnections[$group])) {
                    $buffer = serialize(array());
                    $connection->send(pack('N', strlen($buffer)) . $buffer, true);
                    return;
                }
                $client_info_array = array();
                foreach ($this->_groupConnections[$group] as $connection_id => $client_connection) {
                    $client_info_array[$connection_id] = $client_connection->session;
                }
                $buffer = serialize($client_info_array);
                $connection->send(pack('N', strlen($buffer)) . $buffer, true);
                return;
            // 获取用户组成员数 Gateway::getClientCountByGroup($group);
            case GatewayProtocol::CMD_GET_CLIENT_COUNT_BY_GROUP:
                $group = $data['ext_data'];
                $count = 0;
                if ($group !== '') {
                    if (isset($this->_groupConnections[$group])) {
                        $count = count($this->_groupConnections[$group]);
                    }
                } else {
                    $count = count($this->_clientConnections);
                }
                $buffer = serialize($count);
                $connection->send(pack('N', strlen($buffer)) . $buffer, true);
                return;
            // 获取与某个 uid 绑定的所有 client_id Gateway::getClientIdByUid($uid);
            case GatewayProtocol::CMD_GET_CLIENT_ID_BY_UID:
                $uid = $data['ext_data'];
                if (empty($this->_uidConnections[$uid])) {
                    $buffer = serialize(array());
                } else {
                    $buffer = serialize(array_keys($this->_uidConnections[$uid]));
                }
                $connection->send(pack('N', strlen($buffer)) . $buffer, true);
                return;
            default :
                $err_msg = "gateway inner pack err cmd=$cmd";
                throw new \Exception($err_msg);
        }
    }
914858
public function onMessage($server, $serverSocket, $frame)
    {
        $id = $frame->fd;

        if (isset($this->connections[$id])) {
            $this->fireMessageEvent(
                $this->connections[$id]['handler'],
                $this->connections[$id]['connection'],
                $frame
            );
        }
    }
1147832
public function onSubscribe(ConnectionInterface $connection, Topic $topic, WampRequest $request)
    {
        //this will broadcast the message to ALL subscribers of this topic.
        $topic->broadcast(['msg' => $connection->resourceId . " has joined " . $topic->getId()]);
    }