first commit

This commit is contained in:
浪子小新
2020-03-22 12:20:07 +08:00
commit bf9e9a2dfd
85 changed files with 17363 additions and 0 deletions
@@ -0,0 +1,4 @@
.buildpath
.project
.settings
.idea
@@ -0,0 +1,21 @@
The MIT License
Copyright (c) 2009-2015 walkor<[email protected]> and contributors (see https://github.com/walkor/workerman/contributors)
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
+38
View File
@@ -0,0 +1,38 @@
GatewayWorker
=================
GatewayWorker基于[Workerman](https://github.com/walkor/Workerman)开发的一个项目框架,用于快速开发长连接应用,例如app推送服务端、即时IM服务端、游戏服务端、物联网、智能家居等等。
GatewayWorker使用经典的Gateway和Worker进程模型。Gateway进程负责维持客户端连接,并转发客户端的数据给Worker进程处理;Worker进程负责处理实际的业务逻辑,并将结果推送给对应的客户端。Gateway服务和Worker服务可以分开部署在不同的服务器上,实现分布式集群。
GatewayWorker提供非常方便的API,可以全局广播数据、可以向某个群体广播数据、也可以向某个特定客户端推送数据。配合Workerman的定时器,也可以定时推送数据。
快速开始
======
开发者可以从一个简单的demo开始(demo中包含了GatewayWorker内核,以及start_gateway.php start_business.php等启动入口文件)<br>
[点击这里下载demo](http://www.workerman.net/download/GatewayWorker.zip)。<br>
demo说明见源码readme。
手册
=======
http://www.workerman.net/gatewaydoc/
安装内核
=======
只安装GatewayWorker内核文件(不包含start_gateway.php start_businessworker.php等启动入口文件)
```
composer require workerman/gateway-worker
```
使用GatewayWorker开发的项目
=======
## [tadpole](http://kedou.workerman.net/)
[Live demo](http://kedou.workerman.net/)
[Source code](https://github.com/walkor/workerman)
![workerman todpole](http://www.workerman.net/img/workerman-todpole.png)
## [chat room](http://chat.workerman.net/)
[Live demo](http://chat.workerman.net/)
[Source code](https://github.com/walkor/workerman-chat)
![workerman-chat](http://www.workerman.net/img/workerman-chat.png)
@@ -0,0 +1,12 @@
{
"name" : "workerman/gateway-worker",
"keywords": ["distributed","communication"],
"homepage": "http://www.workerman.net",
"license" : "MIT",
"require": {
"workerman/workerman" : ">=3.1.8"
},
"autoload": {
"psr-4": {"GatewayWorker\\": "./src"}
}
}
@@ -0,0 +1,560 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<[email protected]>
* @copyright walkor<[email protected]>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace GatewayWorker;
use Workerman\Connection\TcpConnection;
use Workerman\Worker;
use Workerman\Lib\Timer;
use Workerman\Connection\AsyncTcpConnection;
use GatewayWorker\Protocols\GatewayProtocol;
use GatewayWorker\Lib\Context;
/**
*
* BusinessWorker 用于处理Gateway转发来的数据
*
* @author walkor<[email protected]>
*
*/
class BusinessWorker extends Worker
{
/**
* 保存与 gateway 的连接 connection 对象
*
* @var array
*/
public $gatewayConnections = array();
/**
* 注册中心地址
*
* @var string|array
*/
public $registerAddress = '127.0.0.1:1236';
/**
* 事件处理类,默认是 Event 类
*
* @var string
*/
public $eventHandler = 'Events';
/**
* 业务超时时间,可用来定位程序卡在哪里
*
* @var int
*/
public $processTimeout = 30;
/**
* 业务超时时间,可用来定位程序卡在哪里
*
* @var callable
*/
public $processTimeoutHandler = '\\Workerman\\Worker::log';
/**
* 秘钥
*
* @var string
*/
public $secretKey = '';
/**
* businessWorker进程将消息转发给gateway进程的发送缓冲区大小
*
* @var int
*/
public $sendToGatewayBufferSize = 10240000;
/**
* 保存用户设置的 worker 启动回调
*
* @var callback
*/
protected $_onWorkerStart = null;
/**
* 保存用户设置的 workerReload 回调
*
* @var callback
*/
protected $_onWorkerReload = null;
/**
* 保存用户设置的 workerStop 回调
*
* @var callback
*/
protected $_onWorkerStop= null;
/**
* 到注册中心的连接
*
* @var AsyncTcpConnection
*/
protected $_registerConnection = null;
/**
* 处于连接状态的 gateway 通讯地址
*
* @var array
*/
protected $_connectingGatewayAddresses = array();
/**
* 所有 geteway 内部通讯地址
*
* @var array
*/
protected $_gatewayAddresses = array();
/**
* 等待连接个 gateway 地址
*
* @var array
*/
protected $_waitingConnectGatewayAddresses = array();
/**
* Event::onConnect 回调
*
* @var callback
*/
protected $_eventOnConnect = null;
/**
* Event::onMessage 回调
*
* @var callback
*/
protected $_eventOnMessage = null;
/**
* Event::onClose 回调
*
* @var callback
*/
protected $_eventOnClose = null;
/**
* websocket回调
*
* @var null
*/
protected $_eventOnWebSocketConnect = null;
/**
* SESSION 版本缓存
*
* @var array
*/
protected $_sessionVersion = array();
/**
* 用于保持长连接的心跳时间间隔
*
* @var int
*/
const PERSISTENCE_CONNECTION_PING_INTERVAL = 25;
/**
* 构造函数
*
* @param string $socket_name
* @param array $context_option
*/
public function __construct($socket_name = '', $context_option = array())
{
parent::__construct($socket_name, $context_option);
$backrace = debug_backtrace();
$this->_autoloadRootPath = dirname($backrace[0]['file']);
}
/**
* {@inheritdoc}
*/
public function run()
{
$this->_onWorkerStart = $this->onWorkerStart;
$this->_onWorkerReload = $this->onWorkerReload;
$this->_onWorkerStop = $this->onWorkerStop;
$this->onWorkerStop = array($this, 'onWorkerStop');
$this->onWorkerStart = array($this, 'onWorkerStart');
$this->onWorkerReload = array($this, 'onWorkerReload');
parent::run();
}
/**
* 当进程启动时一些初始化工作
*
* @return void
*/
protected function onWorkerStart()
{
if (!class_exists('\Protocols\GatewayProtocol')) {
class_alias('GatewayWorker\Protocols\GatewayProtocol', 'Protocols\GatewayProtocol');
}
if (!is_array($this->registerAddress)) {
$this->registerAddress = array($this->registerAddress);
}
$this->connectToRegister();
\GatewayWorker\Lib\Gateway::setBusinessWorker($this);
\GatewayWorker\Lib\Gateway::$secretKey = $this->secretKey;
if ($this->_onWorkerStart) {
call_user_func($this->_onWorkerStart, $this);
}
if (is_callable($this->eventHandler . '::onWorkerStart')) {
call_user_func($this->eventHandler . '::onWorkerStart', $this);
}
if (function_exists('pcntl_signal')) {
// 业务超时信号处理
pcntl_signal(SIGALRM, array($this, 'timeoutHandler'), false);
} else {
$this->processTimeout = 0;
}
// 设置回调
if (is_callable($this->eventHandler . '::onConnect')) {
$this->_eventOnConnect = $this->eventHandler . '::onConnect';
}
if (is_callable($this->eventHandler . '::onMessage')) {
$this->_eventOnMessage = $this->eventHandler . '::onMessage';
} else {
echo "Waring: {$this->eventHandler}::onMessage is not callable\n";
}
if (is_callable($this->eventHandler . '::onClose')) {
$this->_eventOnClose = $this->eventHandler . '::onClose';
}
if (is_callable($this->eventHandler . '::onWebSocketConnect')) {
$this->_eventOnWebSocketConnect = $this->eventHandler . '::onWebSocketConnect';
}
}
/**
* onWorkerReload 回调
*
* @param Worker $worker
*/
protected function onWorkerReload($worker)
{
// 防止进程立刻退出
$worker->reloadable = false;
// 延迟 0.05 秒退出,避免 BusinessWorker 瞬间全部退出导致没有可用的 BusinessWorker 进程
Timer::add(0.05, array('Workerman\Worker', 'stopAll'));
// 执行用户定义的 onWorkerReload 回调
if ($this->_onWorkerReload) {
call_user_func($this->_onWorkerReload, $this);
}
}
/**
* 当进程关闭时一些清理工作
*
* @return void
*/
protected function onWorkerStop()
{
if ($this->_onWorkerStop) {
call_user_func($this->_onWorkerStop, $this);
}
if (is_callable($this->eventHandler . '::onWorkerStop')) {
call_user_func($this->eventHandler . '::onWorkerStop', $this);
}
}
/**
* 连接服务注册中心
*
* @return void
*/
public function connectToRegister()
{
foreach ($this->registerAddress as $register_address) {
$register_connection = new AsyncTcpConnection("text://{$register_address}");
$secret_key = $this->secretKey;
$register_connection->onConnect = function () use ($register_connection, $secret_key, $register_address) {
$register_connection->send('{"event":"worker_connect","secret_key":"' . $secret_key . '"}');
// 如果Register服务器不在本地服务器,则需要保持心跳
if (strpos($register_address, '127.0.0.1') !== 0) {
$register_connection->ping_timer = Timer::add(self::PERSISTENCE_CONNECTION_PING_INTERVAL, function () use ($register_connection) {
$register_connection->send('{"event":"ping"}');
});
}
};
$register_connection->onClose = function ($register_connection) {
if(!empty($register_connection->ping_timer)) {
Timer::del($register_connection->ping_timer);
}
$register_connection->reconnect(1);
};
$register_connection->onMessage = array($this, 'onRegisterConnectionMessage');
$register_connection->connect();
}
}
/**
* 当注册中心发来消息时
*
* @return void
*/
public function onRegisterConnectionMessage($register_connection, $data)
{
$data = json_decode($data, true);
if (!isset($data['event'])) {
echo "Received bad data from Register\n";
return;
}
$event = $data['event'];
switch ($event) {
case 'broadcast_addresses':
if (!is_array($data['addresses'])) {
echo "Received bad data from Register. Addresses empty\n";
return;
}
$addresses = $data['addresses'];
$this->_gatewayAddresses = array();
foreach ($addresses as $addr) {
$this->_gatewayAddresses[$addr] = $addr;
}
$this->checkGatewayConnections($addresses);
break;
default:
echo "Receive bad event:$event from Register.\n";
}
}
/**
* 当 gateway 转发来数据时
*
* @param TcpConnection $connection
* @param mixed $data
*/
public function onGatewayMessage($connection, $data)
{
$cmd = $data['cmd'];
if ($cmd === GatewayProtocol::CMD_PING) {
return;
}
// 上下文数据
Context::$client_ip = $data['client_ip'];
Context::$client_port = $data['client_port'];
Context::$local_ip = $data['local_ip'];
Context::$local_port = $data['local_port'];
Context::$connection_id = $data['connection_id'];
Context::$client_id = Context::addressToClientId($data['local_ip'], $data['local_port'],
$data['connection_id']);
// $_SERVER 变量
$_SERVER = array(
'REMOTE_ADDR' => long2ip($data['client_ip']),
'REMOTE_PORT' => $data['client_port'],
'GATEWAY_ADDR' => long2ip($data['local_ip']),
'GATEWAY_PORT' => $data['gateway_port'],
'GATEWAY_CLIENT_ID' => Context::$client_id,
);
// 检查session版本,如果是过期的session数据则拉取最新的数据
if ($cmd !== GatewayProtocol::CMD_ON_CLOSE && isset($this->_sessionVersion[Context::$client_id]) && $this->_sessionVersion[Context::$client_id] !== crc32($data['ext_data'])) {
$_SESSION = Context::$old_session = \GatewayWorker\Lib\Gateway::getSession(Context::$client_id);
} else {
if (!isset($this->_sessionVersion[Context::$client_id])) {
$this->_sessionVersion[Context::$client_id] = crc32($data['ext_data']);
}
// 尝试解析 session
if ($data['ext_data'] != '') {
Context::$old_session = $_SESSION = Context::sessionDecode($data['ext_data']);
} else {
Context::$old_session = $_SESSION = null;
}
}
if ($this->processTimeout) {
pcntl_alarm($this->processTimeout);
}
// 尝试执行 Event::onConnection、Event::onMessage、Event::onClose
switch ($cmd) {
case GatewayProtocol::CMD_ON_CONNECT:
if ($this->_eventOnConnect) {
call_user_func($this->_eventOnConnect, Context::$client_id);
}
break;
case GatewayProtocol::CMD_ON_MESSAGE:
if ($this->_eventOnMessage) {
call_user_func($this->_eventOnMessage, Context::$client_id, $data['body']);
}
break;
case GatewayProtocol::CMD_ON_CLOSE:
unset($this->_sessionVersion[Context::$client_id]);
if ($this->_eventOnClose) {
call_user_func($this->_eventOnClose, Context::$client_id);
}
break;
case GatewayProtocol::CMD_ON_WEBSOCKET_CONNECT:
if ($this->_eventOnWebSocketConnect) {
call_user_func($this->_eventOnWebSocketConnect, Context::$client_id, $data['body']);
}
break;
}
if ($this->processTimeout) {
pcntl_alarm(0);
}
// session 必须是数组
if ($_SESSION !== null && !is_array($_SESSION)) {
throw new \Exception('$_SESSION must be an array. But $_SESSION=' . var_export($_SESSION, true) . ' is not array.');
}
// 判断 session 是否被更改
if ($_SESSION !== Context::$old_session && $cmd !== GatewayProtocol::CMD_ON_CLOSE) {
$session_str_now = $_SESSION !== null ? Context::sessionEncode($_SESSION) : '';
\GatewayWorker\Lib\Gateway::setSocketSession(Context::$client_id, $session_str_now);
$this->_sessionVersion[Context::$client_id] = crc32($session_str_now);
}
Context::clear();
}
/**
* 当与 Gateway 的连接断开时触发
*
* @param TcpConnection $connection
* @return void
*/
public function onGatewayClose($connection)
{
$addr = $connection->remoteAddress;
unset($this->gatewayConnections[$addr], $this->_connectingGatewayAddresses[$addr]);
if (isset($this->_gatewayAddresses[$addr]) && !isset($this->_waitingConnectGatewayAddresses[$addr])) {
Timer::add(1, array($this, 'tryToConnectGateway'), array($addr), false);
$this->_waitingConnectGatewayAddresses[$addr] = $addr;
}
}
/**
* 尝试连接 Gateway 内部通讯地址
*
* @param string $addr
*/
public function tryToConnectGateway($addr)
{
if (!isset($this->gatewayConnections[$addr]) && !isset($this->_connectingGatewayAddresses[$addr]) && isset($this->_gatewayAddresses[$addr])) {
$gateway_connection = new AsyncTcpConnection("GatewayProtocol://$addr");
$gateway_connection->remoteAddress = $addr;
$gateway_connection->onConnect = array($this, 'onConnectGateway');
$gateway_connection->onMessage = array($this, 'onGatewayMessage');
$gateway_connection->onClose = array($this, 'onGatewayClose');
$gateway_connection->onError = array($this, 'onGatewayError');
$gateway_connection->maxSendBufferSize = $this->sendToGatewayBufferSize;
if (TcpConnection::$defaultMaxSendBufferSize == $gateway_connection->maxSendBufferSize) {
$gateway_connection->maxSendBufferSize = 50 * 1024 * 1024;
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_WORKER_CONNECT;
$gateway_data['body'] = json_encode(array(
'worker_key' =>"{$this->name}:{$this->id}",
'secret_key' => $this->secretKey,
));
$gateway_connection->send($gateway_data);
$gateway_connection->connect();
$this->_connectingGatewayAddresses[$addr] = $addr;
}
unset($this->_waitingConnectGatewayAddresses[$addr]);
}
/**
* 检查 gateway 的通信端口是否都已经连
* 如果有未连接的端口,则尝试连接
*
* @param array $addresses_list
*/
public function checkGatewayConnections($addresses_list)
{
if (empty($addresses_list)) {
return;
}
foreach ($addresses_list as $addr) {
if (!isset($this->_waitingConnectGatewayAddresses[$addr])) {
$this->tryToConnectGateway($addr);
}
}
}
/**
* 当连接上 gateway 的通讯端口时触发
* 将连接 connection 对象保存起来
*
* @param TcpConnection $connection
* @return void
*/
public function onConnectGateway($connection)
{
$this->gatewayConnections[$connection->remoteAddress] = $connection;
unset($this->_connectingGatewayAddresses[$connection->remoteAddress], $this->_waitingConnectGatewayAddresses[$connection->remoteAddress]);
}
/**
* 当与 gateway 的连接出现错误时触发
*
* @param TcpConnection $connection
* @param int $error_no
* @param string $error_msg
*/
public function onGatewayError($connection, $error_no, $error_msg)
{
echo "GatewayConnection Error : $error_no ,$error_msg\n";
}
/**
* 获取所有 Gateway 内部通讯地址
*
* @return array
*/
public function getAllGatewayAddresses()
{
return $this->_gatewayAddresses;
}
/**
* 业务超时回调
*
* @param int $signal
* @throws \Exception
*/
public function timeoutHandler($signal)
{
switch ($signal) {
// 超时时钟
case SIGALRM:
// 超时异常
$e = new \Exception("process_timeout", 506);
$trace_str = $e->getTraceAsString();
// 去掉第一行timeoutHandler的调用栈
$trace_str = $e->getMessage() . ":\n" . substr($trace_str, strpos($trace_str, "\n") + 1) . "\n";
// 开发者没有设置超时处理函数,或者超时处理函数返回空则执行退出
if (!$this->processTimeoutHandler || !call_user_func($this->processTimeoutHandler, $trace_str, $e)) {
Worker::stopAll();
}
break;
}
}
}
@@ -0,0 +1,1019 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<[email protected]>
* @copyright walkor<[email protected]>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace GatewayWorker;
use GatewayWorker\Lib\Context;
use Workerman\Connection\TcpConnection;
use Workerman\Worker;
use Workerman\Lib\Timer;
use Workerman\Autoloader;
use Workerman\Connection\AsyncTcpConnection;
use GatewayWorker\Protocols\GatewayProtocol;
/**
*
* Gateway,基于Worker 开发
* 用于转发客户端的数据给Worker处理,以及转发Worker的数据给客户端
*
* @author walkor<[email protected]>
*
*/
class Gateway extends Worker
{
/**
* 版本
*
* @var string
*/
const VERSION = '3.0.13';
/**
* 本机 IP
* 单机部署默认 127.0.0.1,如果是分布式部署,需要设置成本机 IP
*
* @var string
*/
public $lanIp = '127.0.0.1';
/**
* 本机端口
*
* @var string
*/
public $lanPort = 0;
/**
* gateway 内部通讯起始端口,每个 gateway 实例应该都不同,步长1000
*
* @var int
*/
public $startPort = 2000;
/**
* 注册服务地址,用于注册 Gateway BusinessWorker,使之能够通讯
*
* @var string|array
*/
public $registerAddress = '127.0.0.1:1236';
/**
* 是否可以平滑重启,gateway 不能平滑重启,否则会导致连接断开
*
* @var bool
*/
public $reloadable = false;
/**
* 心跳时间间隔
*
* @var int
*/
public $pingInterval = 0;
/**
* $pingNotResponseLimit * $pingInterval 时间内,客户端未发送任何数据,断开客户端连接
*
* @var int
*/
public $pingNotResponseLimit = 0;
/**
* 服务端向客户端发送的心跳数据
*
* @var string
*/
public $pingData = '';
/**
* 秘钥
*
* @var string
*/
public $secretKey = '';
/**
* 路由函数
*
* @var callback
*/
public $router = null;
/**
* gateway进程转发给businessWorker进程的发送缓冲区大小
*
* @var int
*/
public $sendToWorkerBufferSize = 10240000;
/**
* gateway进程将数据发给客户端时每个客户端发送缓冲区大小
*
* @var int
*/
public $sendToClientBufferSize = 1024000;
/**
* 协议加速
*
* @var bool
*/
public $protocolAccelerate = false;
/**
* 保存客户端的所有 connection 对象
*
* @var array
*/
protected $_clientConnections = array();
/**
* uid 到 connection 的映射,一对多关系
*/
protected $_uidConnections = array();
/**
* group 到 connection 的映射,一对多关系
*
* @var array
*/
protected $_groupConnections = array();
/**
* 保存所有 worker 的内部连接的 connection 对象
*
* @var array
*/
protected $_workerConnections = array();
/**
* gateway 内部监听 worker 内部连接的 worker
*
* @var Worker
*/
protected $_innerTcpWorker = null;
/**
* 当 worker 启动时
*
* @var callback
*/
protected $_onWorkerStart = null;
/**
* 当有客户端连接时
*
* @var callback
*/
protected $_onConnect = null;
/**
* 当客户端发来消息时
*
* @var callback
*/
protected $_onMessage = null;
/**
* 当客户端连接关闭时
*
* @var callback
*/
protected $_onClose = null;
/**
* 当 worker 停止时
*
* @var callback
*/
protected $_onWorkerStop = null;
/**
* 进程启动时间
*
* @var int
*/
protected $_startTime = 0;
/**
* gateway 监听的端口
*
* @var int
*/
protected $_gatewayPort = 0;
/**
* connectionId 记录器
* @var int
*/
protected static $_connectionIdRecorder = 0;
/**
* 用于保持长连接的心跳时间间隔
*
* @var int
*/
const PERSISTENCE_CONNECTION_PING_INTERVAL = 25;
/**
* 构造函数
*
* @param string $socket_name
* @param array $context_option
*/
public function __construct($socket_name, $context_option = array())
{
parent::__construct($socket_name, $context_option);
$this->_gatewayPort = substr(strrchr($socket_name,':'),1);
$this->router = array("\\GatewayWorker\\Gateway", 'routerBind');
$backtrace = debug_backtrace();
$this->_autoloadRootPath = dirname($backtrace[0]['file']);
}
/**
* {@inheritdoc}
*/
public function run()
{
// 保存用户的回调,当对应的事件发生时触发
$this->_onWorkerStart = $this->onWorkerStart;
$this->onWorkerStart = array($this, 'onWorkerStart');
// 保存用户的回调,当对应的事件发生时触发
$this->_onConnect = $this->onConnect;
$this->onConnect = array($this, 'onClientConnect');
// onMessage禁止用户设置回调
$this->onMessage = array($this, 'onClientMessage');
// 保存用户的回调,当对应的事件发生时触发
$this->_onClose = $this->onClose;
$this->onClose = array($this, 'onClientClose');
// 保存用户的回调,当对应的事件发生时触发
$this->_onWorkerStop = $this->onWorkerStop;
$this->onWorkerStop = array($this, 'onWorkerStop');
if (!is_array($this->registerAddress)) {
$this->registerAddress = array($this->registerAddress);
}
// 记录进程启动的时间
$this->_startTime = time();
// 运行父方法
parent::run();
}
/**
* 当客户端发来数据时,转发给worker处理
*
* @param TcpConnection $connection
* @param mixed $data
*/
public function onClientMessage($connection, $data)
{
$connection->pingNotResponseCount = -1;
$this->sendToWorker(GatewayProtocol::CMD_ON_MESSAGE, $connection, $data);
}
/**
* 当客户端连接上来时,初始化一些客户端的数据
* 包括全局唯一的client_id、初始化session等
*
* @param TcpConnection $connection
*/
public function onClientConnect($connection)
{
$connection->id = self::generateConnectionId();
// 保存该连接的内部通讯的数据包报头,避免每次重新初始化
$connection->gatewayHeader = array(
'local_ip' => ip2long($this->lanIp),
'local_port' => $this->lanPort,
'client_ip' => ip2long($connection->getRemoteIp()),
'client_port' => $connection->getRemotePort(),
'gateway_port' => $this->_gatewayPort,
'connection_id' => $connection->id,
'flag' => 0,
);
// 连接的 session
$connection->session = '';
// 该连接的心跳参数
$connection->pingNotResponseCount = -1;
// 该链接发送缓冲区大小
$connection->maxSendBufferSize = $this->sendToClientBufferSize;
// 保存客户端连接 connection 对象
$this->_clientConnections[$connection->id] = $connection;
// 如果用户有自定义 onConnect 回调,则执行
if ($this->_onConnect) {
call_user_func($this->_onConnect, $connection);
} elseif ($connection->protocol === '\Workerman\Protocols\Websocket') {
$connection->onWebSocketConnect = array($this, 'onWebsocketConnect');
}
$this->sendToWorker(GatewayProtocol::CMD_ON_CONNECT, $connection);
}
/**
* websocket握手时触发
*
* @param $connection
* @param $http_buffer
*/
public function onWebsocketConnect($connection, $http_buffer)
{
$this->sendToWorker(GatewayProtocol::CMD_ON_WEBSOCKET_CONNECT, $connection, array('get' => $_GET, 'server' => $_SERVER, 'cookie' => $_COOKIE));
}
/**
* 生成connection id
* @return int
*/
protected function generateConnectionId()
{
$max_unsigned_int = 4294967295;
if (self::$_connectionIdRecorder >= $max_unsigned_int) {
self::$_connectionIdRecorder = 0;
}
while(++self::$_connectionIdRecorder <= $max_unsigned_int) {
if(!isset($this->_clientConnections[self::$_connectionIdRecorder])) {
break;
}
}
return self::$_connectionIdRecorder;
}
/**
* 发送数据给 worker 进程
*
* @param int $cmd
* @param TcpConnection $connection
* @param mixed $body
* @return bool
*/
protected function sendToWorker($cmd, $connection, $body = '')
{
$gateway_data = $connection->gatewayHeader;
$gateway_data['cmd'] = $cmd;
$gateway_data['body'] = $body;
$gateway_data['ext_data'] = $connection->session;
if ($this->_workerConnections) {
// 调用路由函数,选择一个worker把请求转发给它
/** @var TcpConnection $worker_connection */
$worker_connection = call_user_func($this->router, $this->_workerConnections, $connection, $cmd, $body);
if (false === $worker_connection->send($gateway_data)) {
$msg = "SendBufferToWorker fail. May be the send buffer are overflow. See http://doc2.workerman.net/send-buffer-overflow.html";
static::log($msg);
return false;
}
} // 没有可用的 worker
else {
// gateway 启动后 1-2 秒内 SendBufferToWorker fail 是正常现象,因为与 worker 的连接还没建立起来,
// 所以不记录日志,只是关闭连接
$time_diff = 2;
if (time() - $this->_startTime >= $time_diff) {
$msg = 'SendBufferToWorker fail. The connections between Gateway and BusinessWorker are not ready. See http://doc2.workerman.net/send-buffer-to-worker-fail.html';
static::log($msg);
}
$connection->destroy();
return false;
}
return true;
}
/**
* 随机路由,返回 worker connection 对象
*
* @param array $worker_connections
* @param TcpConnection $client_connection
* @param int $cmd
* @param mixed $buffer
* @return TcpConnection
*/
public static function routerRand($worker_connections, $client_connection, $cmd, $buffer)
{
return $worker_connections[array_rand($worker_connections)];
}
/**
* client_id 与 worker 绑定
*
* @param array $worker_connections
* @param TcpConnection $client_connection
* @param int $cmd
* @param mixed $buffer
* @return TcpConnection
*/
public static function routerBind($worker_connections, $client_connection, $cmd, $buffer)
{
if (!isset($client_connection->businessworker_address) || !isset($worker_connections[$client_connection->businessworker_address])) {
$client_connection->businessworker_address = array_rand($worker_connections);
}
return $worker_connections[$client_connection->businessworker_address];
}
/**
* 当客户端关闭时
*
* @param TcpConnection $connection
*/
public function onClientClose($connection)
{
// 尝试通知 worker,触发 Event::onClose
$this->sendToWorker(GatewayProtocol::CMD_ON_CLOSE, $connection);
unset($this->_clientConnections[$connection->id]);
// 清理 uid 数据
if (!empty($connection->uid)) {
$uid = $connection->uid;
unset($this->_uidConnections[$uid][$connection->id]);
if (empty($this->_uidConnections[$uid])) {
unset($this->_uidConnections[$uid]);
}
}
// 清理 group 数据
if (!empty($connection->groups)) {
foreach ($connection->groups as $group) {
unset($this->_groupConnections[$group][$connection->id]);
if (empty($this->_groupConnections[$group])) {
unset($this->_groupConnections[$group]);
}
}
}
// 触发 onClose
if ($this->_onClose) {
call_user_func($this->_onClose, $connection);
}
}
/**
* 当 Gateway 启动的时候触发的回调函数
*
* @return void
*/
public function onWorkerStart()
{
// 分配一个内部通讯端口
$this->lanPort = $this->startPort + $this->id;
// 如果有设置心跳,则定时执行
if ($this->pingInterval > 0) {
$timer_interval = $this->pingNotResponseLimit > 0 ? $this->pingInterval / 2 : $this->pingInterval;
Timer::add($timer_interval, array($this, 'ping'));
}
// 如果BusinessWorker ip不是127.0.0.1,则需要加gateway到BusinessWorker的心跳
if ($this->lanIp !== '127.0.0.1') {
Timer::add(self::PERSISTENCE_CONNECTION_PING_INTERVAL, array($this, 'pingBusinessWorker'));
}
if (!class_exists('\Protocols\GatewayProtocol')) {
class_alias('GatewayWorker\Protocols\GatewayProtocol', 'Protocols\GatewayProtocol');
}
// 初始化 gateway 内部的监听,用于监听 worker 的连接已经连接上发来的数据
$this->_innerTcpWorker = new Worker("GatewayProtocol://{$this->lanIp}:{$this->lanPort}");
$this->_innerTcpWorker->listen();
$this->_innerTcpWorker->name = 'GatewayInnerWorker';
// 重新设置自动加载根目录
Autoloader::setRootPath($this->_autoloadRootPath);
// 设置内部监听的相关回调
$this->_innerTcpWorker->onMessage = array($this, 'onWorkerMessage');
$this->_innerTcpWorker->onConnect = array($this, 'onWorkerConnect');
$this->_innerTcpWorker->onClose = array($this, 'onWorkerClose');
// 注册 gateway 的内部通讯地址,worker 去连这个地址,以便 gateway 与 worker 之间建立起 TCP 长连接
$this->registerAddress();
if ($this->_onWorkerStart) {
call_user_func($this->_onWorkerStart, $this);
}
}
/**
* 当 worker 通过内部通讯端口连接到 gateway 时
*
* @param TcpConnection $connection
*/
public function onWorkerConnect($connection)
{
$connection->maxSendBufferSize = $this->sendToWorkerBufferSize;
$connection->authorized = $this->secretKey ? false : true;
}
/**
* 当 worker 发来数据时
*
* @param TcpConnection $connection
* @param mixed $data
* @throws \Exception
*
* @return void
*/
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());
$connection->close();
return;
}
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));
$connection->close();
return;
}
$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));
$connection->close();
return;
}
$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;
case GatewayProtocol::CMD_SELECT:
$client_info_array = array();
$ext_data = json_decode($data['ext_data'], true);
if (!$ext_data) {
echo 'CMD_SELECT ext_data=' . var_export($data['ext_data'], true) . '\r\n';
$buffer = serialize($client_info_array);
$connection->send(pack('N', strlen($buffer)) . $buffer, true);
return;
}
$fields = $ext_data['fields'];
$where = $ext_data['where'];
if ($where) {
$connection_box_map = array(
'groups' => $this->_groupConnections,
'uid' => $this->_uidConnections
);
// $where = ['groups'=>[x,x..], 'uid'=>[x,x..], 'connection_id'=>[x,x..]]
foreach ($where as $key => $items) {
if ($key !== 'connection_id') {
$connections_box = $connection_box_map[$key];
foreach ($items as $item) {
if (isset($connections_box[$item])) {
foreach ($connections_box[$item] as $connection_id => $client_connection) {
if (!isset($client_info_array[$connection_id])) {
$client_info_array[$connection_id] = array();
// $fields = ['groups', 'uid', 'session']
foreach ($fields as $field) {
$client_info_array[$connection_id][$field] = isset($client_connection->$field) ? $client_connection->$field : null;
}
}
}
}
}
} else {
foreach ($items as $connection_id) {
if (isset($this->_clientConnections[$connection_id])) {
$client_connection = $this->_clientConnections[$connection_id];
$client_info_array[$connection_id] = array();
// $fields = ['groups', 'uid', 'session']
foreach ($fields as $field) {
$client_info_array[$connection_id][$field] = isset($client_connection->$field) ? $client_connection->$field : null;
}
}
}
}
}
} else {
foreach ($this->_clientConnections as $connection_id => $client_connection) {
foreach ($fields as $field) {
$client_info_array[$connection_id][$field] = isset($client_connection->$field) ? $client_connection->$field : null;
}
}
}
$buffer = serialize($client_info_array);
$connection->send(pack('N', strlen($buffer)) . $buffer, true);
return;
// 获取在线群组列表
case GatewayProtocol::CMD_GET_GROUP_ID_LIST:
$buffer = serialize(array_keys($this->_groupConnections));
$connection->send(pack('N', strlen($buffer)) . $buffer, true);
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 = array_replace_recursive($session, $session_for_merge);
$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;
// 获得客户端sessions
case GatewayProtocol::CMD_GET_ALL_CLIENT_SESSIONS:
$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]);
if (empty($this->_groupConnections[$group])) {
unset($this->_groupConnections[$group]);
}
return;
// 解散分组
case GatewayProtocol::CMD_UNGROUP:
$group = $data['ext_data'];
if (empty($group)) {
echo "leave(group) group empty, group=" . var_export($group, true);
return;
}
if (empty($this->_groupConnections[$group])) {
return;
}
foreach ($this->_groupConnections[$group] as $client_connection) {
unset($client_connection->groups[$group]);
}
unset($this->_groupConnections[$group]);
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::getClientSessionsByGroup($group);
case GatewayProtocol::CMD_GET_CLIENT_SESSIONS_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";
echo $err_msg;
}
}
/**
* 当worker连接关闭时
*
* @param TcpConnection $connection
*/
public function onWorkerClose($connection)
{
if (isset($connection->key)) {
unset($this->_workerConnections[$connection->key]);
}
}
/**
* 存储当前 Gateway 的内部通信地址
*
* @return bool
*/
public function registerAddress()
{
$address = $this->lanIp . ':' . $this->lanPort;
foreach ($this->registerAddress as $register_address) {
$register_connection = new AsyncTcpConnection("text://{$register_address}");
$secret_key = $this->secretKey;
$register_connection->onConnect = function($register_connection) use ($address, $secret_key, $register_address){
$register_connection->send('{"event":"gateway_connect", "address":"' . $address . '", "secret_key":"' . $secret_key . '"}');
// 如果Register服务器不在本地服务器,则需要保持心跳
if (strpos($register_address, '127.0.0.1') !== 0) {
$register_connection->ping_timer = Timer::add(self::PERSISTENCE_CONNECTION_PING_INTERVAL, function () use ($register_connection) {
$register_connection->send('{"event":"ping"}');
});
}
};
$register_connection->onClose = function ($register_connection) {
if(!empty($register_connection->ping_timer)) {
Timer::del($register_connection->ping_timer);
}
$register_connection->reconnect(1);
};
$register_connection->connect();
}
}
/**
* 心跳逻辑
*
* @return void
*/
public function ping()
{
$ping_data = $this->pingData ? (string)$this->pingData : null;
$raw = false;
if ($this->protocolAccelerate && $ping_data && $this->protocol) {
$ping_data = $this->preEncodeForClient($ping_data);
$raw = true;
}
// 遍历所有客户端连接
foreach ($this->_clientConnections as $connection) {
// 上次发送的心跳还没有回复次数大于限定值就断开
if ($this->pingNotResponseLimit > 0 &&
$connection->pingNotResponseCount >= $this->pingNotResponseLimit * 2
) {
$connection->destroy();
continue;
}
// $connection->pingNotResponseCount 为 -1 说明最近客户端有发来消息,则不给客户端发送心跳
$connection->pingNotResponseCount++;
if ($ping_data) {
if ($connection->pingNotResponseCount === 0 ||
($this->pingNotResponseLimit > 0 && $connection->pingNotResponseCount % 2 === 1)
) {
continue;
}
$connection->send($ping_data, $raw);
}
}
}
/**
* 向 BusinessWorker 发送心跳数据,用于保持长连接
*
* @return void
*/
public function pingBusinessWorker()
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_PING;
foreach ($this->_workerConnections as $connection) {
$connection->send($gateway_data);
}
}
/**
* @param mixed $data
*
* @return string
*/
protected function preEncodeForClient($data)
{
foreach ($this->_clientConnections as $client_connection) {
return call_user_func(array($client_connection->protocol, 'encode'), $data, $client_connection);
}
}
/**
* 当 gateway 关闭时触发,清理数据
*
* @return void
*/
public function onWorkerStop()
{
// 尝试触发用户设置的回调
if ($this->_onWorkerStop) {
call_user_func($this->_onWorkerStop, $this);
}
}
/**
* Log.
* @param string $msg
*/
public static function log($msg){
Timer::add(1, function() use ($msg) {
Worker::log($msg);
}, null, false);
}
}
@@ -0,0 +1,136 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<[email protected]>
* @copyright walkor<[email protected]>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace GatewayWorker\Lib;
use Exception;
/**
* 上下文 包含当前用户 uid 内部通信 local_ip local_port socket_id,以及客户端 client_ip client_port
*/
class Context
{
/**
* 内部通讯 id
*
* @var string
*/
public static $local_ip;
/**
* 内部通讯端口
*
* @var int
*/
public static $local_port;
/**
* 客户端 ip
*
* @var string
*/
public static $client_ip;
/**
* 客户端端口
*
* @var int
*/
public static $client_port;
/**
* client_id
*
* @var string
*/
public static $client_id;
/**
* 连接 connection->id
*
* @var int
*/
public static $connection_id;
/**
* 旧的session
*
* @var string
*/
public static $old_session;
/**
* 编码 session
*
* @param mixed $session_data
* @return string
*/
public static function sessionEncode($session_data = '')
{
if ($session_data !== '') {
return serialize($session_data);
}
return '';
}
/**
* 解码 session
*
* @param string $session_buffer
* @return mixed
*/
public static function sessionDecode($session_buffer)
{
return unserialize($session_buffer);
}
/**
* 清除上下文
*
* @return void
*/
public static function clear()
{
self::$local_ip = self::$local_port = self::$client_ip = self::$client_port =
self::$client_id = self::$connection_id = self::$old_session = null;
}
/**
* 通讯地址到 client_id 的转换
*
* @param int $local_ip
* @param int $local_port
* @param int $connection_id
* @return string
*/
public static function addressToClientId($local_ip, $local_port, $connection_id)
{
return bin2hex(pack('NnN', $local_ip, $local_port, $connection_id));
}
/**
* client_id 到通讯地址的转换
*
* @param string $client_id
* @return array
* @throws Exception
*/
public static function clientIdToAddress($client_id)
{
if (strlen($client_id) !== 20) {
echo new Exception("client_id $client_id is invalid");
return false;
}
return unpack('Nlocal_ip/nlocal_port/Nconnection_id', pack('H*', $client_id));
}
}
@@ -0,0 +1,76 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<[email protected]>
* @copyright walkor<[email protected]>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace GatewayWorker\Lib;
use Config\Db as DbConfig;
use Exception;
/**
* 数据库类
*/
class Db
{
/**
* 实例数组
*
* @var array
*/
protected static $instance = array();
/**
* 获取实例
*
* @param string $config_name
* @return DbConnection
* @throws Exception
*/
public static function instance($config_name)
{
if (!isset(DbConfig::$$config_name)) {
echo "\\Config\\Db::$config_name not set\n";
throw new Exception("\\Config\\Db::$config_name not set\n");
}
if (empty(self::$instance[$config_name])) {
$config = DbConfig::$$config_name;
self::$instance[$config_name] = new DbConnection($config['host'], $config['port'],
$config['user'], $config['password'], $config['dbname'],$config['charset']);
}
return self::$instance[$config_name];
}
/**
* 关闭数据库实例
*
* @param string $config_name
*/
public static function close($config_name)
{
if (isset(self::$instance[$config_name])) {
self::$instance[$config_name]->closeConnection();
self::$instance[$config_name] = null;
}
}
/**
* 关闭所有数据库实例
*/
public static function closeAll()
{
foreach (self::$instance as $connection) {
$connection->closeConnection();
}
self::$instance = array();
}
}
@@ -0,0 +1,1976 @@
<?php
namespace GatewayWorker\Lib;
use Exception;
use PDO;
use PDOException;
/**
* 数据库连接类,依赖 PDO_MYSQL 扩展
* 在 https://github.com/auraphp/Aura.SqlQuery 的基础上修改而成
*/
class DbConnection
{
/**
* SELECT
*
* @var array
*/
protected $union = array();
/**
* 是否是更新
*
* @var bool
*/
protected $for_update = false;
/**
* 选择的列
*
* @var array
*/
protected $cols = array();
/**
* 从哪些表里面 SELECT
*
* @var array
*/
protected $from = array();
/**
* $from 当前的 key
*
* @var int
*/
protected $from_key = -1;
/**
* GROUP BY 的列
*
* @var array
*/
protected $group_by = array();
/**
* HAVING 条件数组.
*
* @var array
*/
protected $having = array();
/**
* HAVING 语句中绑定的值.
*
* @var array
*/
protected $bind_having = array();
/**
* 每页多少条记录
*
* @var int
*/
protected $paging = 10;
/**
* sql 中绑定的值
*
* @var array
*/
protected $bind_values = array();
/**
* WHERE 条件.
*
* @var array
*/
protected $where = array();
/**
* WHERE 语句绑定的值
*
* @var array
*/
protected $bind_where = array();
/**
* ORDER BY 的列
*
* @var array
*/
protected $order_by = array();
/**
* ORDER BY 的排序方式,默认为升序
*
* @var bool
*/
protected $order_asc = true;
/**
* SELECT 多少记录
*
* @var int
*/
protected $limit = 0;
/**
* 返回记录的游标
*
* @var int
*/
protected $offset = 0;
/**
* flags 列表
*
* @var array
*/
protected $flags = array();
/**
* 操作哪个表
*
* @var string
*/
protected $table;
/**
* 表.列 和 last-insert-id 映射
*
* @var array
*/
protected $last_insert_id_names = array();
/**
* INSERT 或者 UPDATE 的列
*
* @param array
*/
protected $col_values;
/**
* 返回的列
*
* @var array
*/
protected $returning = array();
/**
* sql 的类型 SELECT INSERT DELETE UPDATE
*
* @var string
*/
protected $type = '';
/**
* pdo 实例
*
* @var PDO
*/
protected $pdo;
/**
* PDOStatement 实例
*
* @var \PDOStatement
*/
protected $sQuery;
/**
* 数据库用户名密码等配置
*
* @var array
*/
protected $settings = array();
/**
* sql 的参数
*
* @var array
*/
protected $parameters = array();
/**
* 最后一条直行的 sql
*
* @var string
*/
protected $lastSql = '';
/**
* 是否执行成功
*
* @var bool
*/
protected $success = false;
/**
* 选择哪些列
*
* @param string|array $cols
* @return self
*/
public function select($cols = '*')
{
$this->type = 'SELECT';
if (!is_array($cols)) {
$cols = array($cols);
}
$this->cols($cols);
return $this;
}
/**
* 从哪个表删除
*
* @param string $table
* @return self
*/
public function delete($table)
{
$this->type = 'DELETE';
$this->table = $this->quoteName($table);
$this->fromRaw($this->quoteName($table));
return $this;
}
/**
* 更新哪个表
*
* @param string $table
* @return self
*/
public function update($table)
{
$this->type = 'UPDATE';
$this->table = $this->quoteName($table);
return $this;
}
/**
* 向哪个表插入
*
* @param string $table
* @return self
*/
public function insert($table)
{
$this->type = 'INSERT';
$this->table = $this->quoteName($table);
return $this;
}
/**
*
* 设置 SQL_CALC_FOUND_ROWS 标记.
*
* @param bool $enable
* @return self
*/
public function calcFoundRows($enable = true)
{
$this->setFlag('SQL_CALC_FOUND_ROWS', $enable);
return $this;
}
/**
* 设置 SQL_CACHE 标记
*
* @param bool $enable
* @return self
*/
public function cache($enable = true)
{
$this->setFlag('SQL_CACHE', $enable);
return $this;
}
/**
* 设置 SQL_NO_CACHE 标记
*
* @param bool $enable
* @return self
*/
public function noCache($enable = true)
{
$this->setFlag('SQL_NO_CACHE', $enable);
return $this;
}
/**
* 设置 STRAIGHT_JOIN 标记.
*
* @param bool $enable
* @return self
*/
public function straightJoin($enable = true)
{
$this->setFlag('STRAIGHT_JOIN', $enable);
return $this;
}
/**
* 设置 HIGH_PRIORITY 标记
*
* @param bool $enable
* @return self
*/
public function highPriority($enable = true)
{
$this->setFlag('HIGH_PRIORITY', $enable);
return $this;
}
/**
* 设置 SQL_SMALL_RESULT 标记
*
* @param bool $enable
* @return self
*/
public function smallResult($enable = true)
{
$this->setFlag('SQL_SMALL_RESULT', $enable);
return $this;
}
/**
* 设置 SQL_BIG_RESULT 标记
*
* @param bool $enable
* @return self
*/
public function bigResult($enable = true)
{
$this->setFlag('SQL_BIG_RESULT', $enable);
return $this;
}
/**
* 设置 SQL_BUFFER_RESULT 标记
*
* @param bool $enable
* @return self
*/
public function bufferResult($enable = true)
{
$this->setFlag('SQL_BUFFER_RESULT', $enable);
return $this;
}
/**
* 设置 FOR UPDATE 标记
*
* @param bool $enable
* @return self
*/
public function forUpdate($enable = true)
{
$this->for_update = (bool)$enable;
return $this;
}
/**
* 设置 DISTINCT 标记
*
* @param bool $enable
* @return self
*/
public function distinct($enable = true)
{
$this->setFlag('DISTINCT', $enable);
return $this;
}
/**
* 设置 LOW_PRIORITY 标记
*
* @param bool $enable
* @return self
*/
public function lowPriority($enable = true)
{
$this->setFlag('LOW_PRIORITY', $enable);
return $this;
}
/**
* 设置 IGNORE 标记
*
* @param bool $enable
* @return self
*/
public function ignore($enable = true)
{
$this->setFlag('IGNORE', $enable);
return $this;
}
/**
* 设置 QUICK 标记
*
* @param bool $enable
* @return self
*/
public function quick($enable = true)
{
$this->setFlag('QUICK', $enable);
return $this;
}
/**
* 设置 DELAYED 标记
*
* @param bool $enable
* @return self
*/
public function delayed($enable = true)
{
$this->setFlag('DELAYED', $enable);
return $this;
}
/**
* 序列化
*
* @return string
*/
public function __toString()
{
$union = '';
if ($this->union) {
$union = implode(' ', $this->union) . ' ';
}
return $union . $this->build();
}
/**
* 设置每页多少条记录
*
* @param int $paging
* @return self
*/
public function setPaging($paging)
{
$this->paging = (int)$paging;
return $this;
}
/**
* 获取每页多少条记录
*
* @return int
*/
public function getPaging()
{
return $this->paging;
}
/**
* 获取绑定在占位符上的值
*/
public function getBindValues()
{
switch ($this->type) {
case 'SELECT':
return $this->getBindValuesSELECT();
case 'DELETE':
case 'UPDATE':
case 'INSERT':
return $this->getBindValuesCOMMON();
default :
throw new Exception("type err");
}
}
/**
* 获取绑定在占位符上的值
*
* @return array
*/
public function getBindValuesSELECT()
{
$bind_values = $this->bind_values;
$i = 1;
foreach ($this->bind_where as $val) {
$bind_values[$i] = $val;
$i++;
}
foreach ($this->bind_having as $val) {
$bind_values[$i] = $val;
$i++;
}
return $bind_values;
}
/**
*
* SELECT选择哪些列
*
* @param mixed $key
* @param string $val
* @return void
*/
protected function addColSELECT($key, $val)
{
if (is_string($key)) {
$this->cols[$val] = $key;
} else {
$this->addColWithAlias($val);
}
}
/**
* SELECT 增加选择的列
*
* @param string $spec
*/
protected function addColWithAlias($spec)
{
$parts = explode(' ', $spec);
$count = count($parts);
if ($count == 2) {
$this->cols[$parts[1]] = $parts[0];
} elseif ($count == 3 && strtoupper($parts[1]) == 'AS') {
$this->cols[$parts[2]] = $parts[0];
} else {
$this->cols[] = $spec;
}
}
/**
* from 哪个表
*
* @param string $table
* @return self
*/
public function from($table)
{
return $this->fromRaw($this->quoteName($table));
}
/**
* from的表
*
* @param string $table
* @return self
*/
public function fromRaw($table)
{
$this->from[] = array($table);
$this->from_key++;
return $this;
}
/**
*
* 子查询
*
* @param string $table
* @param string $name The alias name for the sub-select.
* @return self
*/
public function fromSubSelect($table, $name)
{
$this->from[] = array("($table) AS " . $this->quoteName($name));
$this->from_key++;
return $this;
}
/**
* 增加 join 语句
*
* @param string $table
* @param string $cond
* @param string $type
* @return self
* @throws Exception
*/
public function join($table, $cond = null, $type = '')
{
return $this->joinInternal($type, $table, $cond);
}
/**
* 增加 join 语句
*
* @param string $join inner, left, natural
* @param string $table
* @param string $cond
* @return self
* @throws Exception
*/
protected function joinInternal($join, $table, $cond = null)
{
if (!$this->from) {
throw new Exception('Cannot join() without from()');
}
$join = strtoupper(ltrim("$join JOIN"));
$table = $this->quoteName($table);
$cond = $this->fixJoinCondition($cond);
$this->from[$this->from_key][] = rtrim("$join $table $cond");
return $this;
}
/**
* quote
*
* @param string $cond
* @return string
*
*/
protected function fixJoinCondition($cond)
{
if (!$cond) {
return '';
}
$cond = $this->quoteNamesIn($cond);
if (strtoupper(substr(ltrim($cond), 0, 3)) == 'ON ') {
return $cond;
}
if (strtoupper(substr(ltrim($cond), 0, 6)) == 'USING ') {
return $cond;
}
return 'ON ' . $cond;
}
/**
* inner join
*
* @param string $table
* @param string $cond
* @return self
* @throws Exception
*/
public function innerJoin($table, $cond = null)
{
return $this->joinInternal('INNER', $table, $cond);
}
/**
* left join
*
* @param string $table
* @param string $cond
* @return self
* @throws Exception
*/
public function leftJoin($table, $cond = null)
{
return $this->joinInternal('LEFT', $table, $cond);
}
/**
* right join
*
* @param string $table
* @param string $cond
* @return self
* @throws Exception
*/
public function rightJoin($table, $cond = null)
{
return $this->joinInternal('RIGHT', $table, $cond);
}
/**
* joinSubSelect
*
* @param string $join inner, left, natural
* @param string $spec
* @param string $name sub-select 的别名
* @param string $cond
* @return self
* @throws Exception
*/
public function joinSubSelect($join, $spec, $name, $cond = null)
{
if (!$this->from) {
throw new \Exception('Cannot join() without from() first.');
}
$join = strtoupper(ltrim("$join JOIN"));
$name = $this->quoteName($name);
$cond = $this->fixJoinCondition($cond);
$this->from[$this->from_key][] = rtrim("$join ($spec) AS $name $cond");
return $this;
}
/**
* group by 语句
*
* @param array $cols
* @return self
*/
public function groupBy(array $cols)
{
foreach ($cols as $col) {
$this->group_by[] = $this->quoteNamesIn($col);
}
return $this;
}
/**
* having 语句
*
* @param string $cond
* @return self
*/
public function having($cond)
{
$this->addClauseCondWithBind('having', 'AND', func_get_args());
return $this;
}
/**
* or having 语句
*
* @param string $cond The HAVING condition.
* @return self
*/
public function orHaving($cond)
{
$this->addClauseCondWithBind('having', 'OR', func_get_args());
return $this;
}
/**
* 设置每页的记录数量
*
* @param int $page
* @return self
*/
public function page($page)
{
$this->limit = 0;
$this->offset = 0;
$page = (int)$page;
if ($page > 0) {
$this->limit = $this->paging;
$this->offset = $this->paging * ($page - 1);
}
return $this;
}
/**
* union
*
* @return self
*/
public function union()
{
$this->union[] = $this->build() . ' UNION';
$this->reset();
return $this;
}
/**
* unionAll
*
* @return self
*/
public function unionAll()
{
$this->union[] = $this->build() . ' UNION ALL';
$this->reset();
return $this;
}
/**
* 重置
*/
protected function reset()
{
$this->resetFlags();
$this->cols = array();
$this->from = array();
$this->from_key = -1;
$this->where = array();
$this->group_by = array();
$this->having = array();
$this->order_by = array();
$this->limit = 0;
$this->offset = 0;
$this->for_update = false;
}
/**
* 清除所有数据
*/
protected function resetAll()
{
$this->union = array();
$this->for_update = false;
$this->cols = array();
$this->from = array();
$this->from_key = -1;
$this->group_by = array();
$this->having = array();
$this->bind_having = array();
$this->paging = 10;
$this->bind_values = array();
$this->where = array();
$this->bind_where = array();
$this->order_by = array();
$this->limit = 0;
$this->offset = 0;
$this->flags = array();
$this->table = '';
$this->last_insert_id_names = array();
$this->col_values = array();
$this->returning = array();
$this->parameters = array();
}
/**
* 创建 SELECT SQL
*
* @return string
*/
protected function buildSELECT()
{
return 'SELECT'
. $this->buildFlags()
. $this->buildCols()
. $this->buildFrom()
. $this->buildWhere()
. $this->buildGroupBy()
. $this->buildHaving()
. $this->buildOrderBy()
. $this->buildLimit()
. $this->buildForUpdate();
}
/**
* 创建 DELETE SQL
*/
protected function buildDELETE()
{
return 'DELETE'
. $this->buildFlags()
. $this->buildFrom()
. $this->buildWhere()
. $this->buildOrderBy()
. $this->buildLimit()
. $this->buildReturning();
}
/**
* 生成 SELECT 列语句
*
* @return string
* @throws Exception
*/
protected function buildCols()
{
if (!$this->cols) {
throw new Exception('No columns in the SELECT.');
}
$cols = array();
foreach ($this->cols as $key => $val) {
if (is_int($key)) {
$cols[] = $this->quoteNamesIn($val);
} else {
$cols[] = $this->quoteNamesIn("$val AS $key");
}
}
return $this->indentCsv($cols);
}
/**
* 生成 FROM 语句.
*
* @return string
*/
protected function buildFrom()
{
if (!$this->from) {
return '';
}
$refs = array();
foreach ($this->from as $from) {
$refs[] = implode(' ', $from);
}
return ' FROM' . $this->indentCsv($refs);
}
/**
* 生成 GROUP BY 语句.
*
* @return string
*/
protected function buildGroupBy()
{
if (!$this->group_by) {
return '';
}
return ' GROUP BY' . $this->indentCsv($this->group_by);
}
/**
* 生成 HAVING 语句.
*
* @return string
*/
protected function buildHaving()
{
if (!$this->having) {
return '';
}
return ' HAVING' . $this->indent($this->having);
}
/**
* 生成 FOR UPDATE 语句
*
* @return string
*/
protected function buildForUpdate()
{
if (!$this->for_update) {
return '';
}
return ' FOR UPDATE';
}
/**
* where
*
* @param string|array $cond
* @return self
*/
public function where($cond)
{
if (is_array($cond)) {
foreach ($cond as $key => $val) {
if (is_string($key)) {
$this->addWhere('AND', array($key, $val));
} else {
$this->addWhere('AND', array($val));
}
}
} else {
$this->addWhere('AND', func_get_args());
}
return $this;
}
/**
* or where
*
* @param string|array $cond
* @return self
*/
public function orWhere($cond)
{
if (is_array($cond)) {
foreach ($cond as $key => $val) {
if (is_string($key)) {
$this->addWhere('OR', array($key, $val));
} else {
$this->addWhere('OR', array($val));
}
}
} else {
$this->addWhere('OR', func_get_args());
}
return $this;
}
/**
* limit
*
* @param int $limit
* @return self
*/
public function limit($limit)
{
$this->limit = (int)$limit;
return $this;
}
/**
* limit offset
*
* @param int $offset
* @return self
*/
public function offset($offset)
{
$this->offset = (int)$offset;
return $this;
}
/**
* orderby.
*
* @param array $cols
* @return self
*/
public function orderBy(array $cols)
{
return $this->addOrderBy($cols);
}
/**
* order by ASC OR DESC
*
* @param array $cols
* @param bool $order_asc
* @return self
*/
public function orderByASC(array $cols, $order_asc = true)
{
$this->order_asc = $order_asc;
return $this->addOrderBy($cols);
}
/**
* order by DESC
*
* @param array $cols
* @return self
*/
public function orderByDESC(array $cols)
{
$this->order_asc = false;
return $this->addOrderBy($cols);
}
// -------------abstractquery----------
/**
* 返回逗号分隔的字符串
*
* @param array $list
* @return string
*/
protected function indentCsv(array $list)
{
return ' ' . implode(',', $list);
}
/**
* 返回空格分隔的字符串
*
* @param array $list
* @return string
*/
protected function indent(array $list)
{
return ' ' . implode(' ', $list);
}
/**
* 批量为占位符绑定值
*
* @param array $bind_values
* @return self
*
*/
public function bindValues(array $bind_values)
{
foreach ($bind_values as $key => $val) {
$this->bindValue($key, $val);
}
return $this;
}
/**
* 单个为占位符绑定值
*
* @param string $name
* @param mixed $value
* @return self
*/
public function bindValue($name, $value)
{
$this->bind_values[$name] = $value;
return $this;
}
/**
* 生成 flag
*
* @return string
*/
protected function buildFlags()
{
if (!$this->flags) {
return '';
}
return ' ' . implode(' ', array_keys($this->flags));
}
/**
* 设置 flag.
*
* @param string $flag
* @param bool $enable
*/
protected function setFlag($flag, $enable = true)
{
if ($enable) {
$this->flags[$flag] = true;
} else {
unset($this->flags[$flag]);
}
}
/**
* 重置 flag
*/
protected function resetFlags()
{
$this->flags = array();
}
/**
*
* 添加 where 语句
*
* @param string $andor 'AND' or 'OR
* @param array $conditions
* @return self
*
*/
protected function addWhere($andor, $conditions)
{
$this->addClauseCondWithBind('where', $andor, $conditions);
return $this;
}
/**
* 添加条件和绑定值
*
* @param string $clause where 、having等
* @param string $andor AND、OR等
* @param array $conditions
*/
protected function addClauseCondWithBind($clause, $andor, $conditions)
{
$cond = array_shift($conditions);
$cond = $this->quoteNamesIn($cond);
$bind =& $this->{"bind_{$clause}"};
foreach ($conditions as $value) {
$bind[] = $value;
}
$clause =& $this->$clause;
if ($clause) {
$clause[] = "$andor $cond";
} else {
$clause[] = $cond;
}
}
/**
* 生成 where 语句
*
* @return string
*/
protected function buildWhere()
{
if (!$this->where) {
return '';
}
return ' WHERE' . $this->indent($this->where);
}
/**
* 增加 order by
*
* @param array $spec The columns and direction to order by.
* @return self
*/
protected function addOrderBy(array $spec)
{
foreach ($spec as $col) {
$this->order_by[] = $this->quoteNamesIn($col);
}
return $this;
}
/**
* 生成 order by 语句
*
* @return string
*/
protected function buildOrderBy()
{
if (!$this->order_by) {
return '';
}
if ($this->order_asc) {
return ' ORDER BY' . $this->indentCsv($this->order_by) . ' ASC';
} else {
return ' ORDER BY' . $this->indentCsv($this->order_by) . ' DESC';
}
}
/**
* 生成 limit 语句
*
* @return string
*/
protected function buildLimit()
{
$has_limit = $this->type == 'DELETE' || $this->type == 'UPDATE';
$has_offset = $this->type == 'SELECT';
if ($has_offset && $this->limit) {
$clause = " LIMIT {$this->limit}";
if ($this->offset) {
$clause .= " OFFSET {$this->offset}";
}
return $clause;
} elseif ($has_limit && $this->limit) {
return " LIMIT {$this->limit}";
}
return '';
}
/**
* Quotes
*
* @param string $spec
* @return string|array
*/
public function quoteName($spec)
{
$spec = trim($spec);
$seps = array(' AS ', ' ', '.');
foreach ($seps as $sep) {
$pos = strripos($spec, $sep);
if ($pos) {
return $this->quoteNameWithSeparator($spec, $sep, $pos);
}
}
return $this->replaceName($spec);
}
/**
* 指定分隔符的 Quotes
*
* @param string $spec
* @param string $sep
* @param int $pos
* @return string
*/
protected function quoteNameWithSeparator($spec, $sep, $pos)
{
$len = strlen($sep);
$part1 = $this->quoteName(substr($spec, 0, $pos));
$part2 = $this->replaceName(substr($spec, $pos + $len));
return "{$part1}{$sep}{$part2}";
}
/**
* Quotes "table.col" 格式的字符串
*
* @param string $text
* @return string|array
*/
public function quoteNamesIn($text)
{
$list = $this->getListForQuoteNamesIn($text);
$last = count($list) - 1;
$text = null;
foreach ($list as $key => $val) {
if (($key + 1) % 3) {
$text .= $this->quoteNamesInLoop($val, $key == $last);
}
}
return $text;
}
/**
* 返回 quote 元素列表
*
* @param string $text
* @return array
*/
protected function getListForQuoteNamesIn($text)
{
$apos = "'";
$quot = '"';
return preg_split(
"/(($apos+|$quot+|\\$apos+|\\$quot+).*?\\2)/",
$text,
-1,
PREG_SPLIT_DELIM_CAPTURE
);
}
/**
* 循环 quote
*
* @param string $val
* @param bool $is_last
* @return string
*/
protected function quoteNamesInLoop($val, $is_last)
{
if ($is_last) {
return $this->replaceNamesAndAliasIn($val);
}
return $this->replaceNamesIn($val);
}
/**
* 替换成别名
*
* @param string $val
* @return string
*/
protected function replaceNamesAndAliasIn($val)
{
$quoted = $this->replaceNamesIn($val);
$pos = strripos($quoted, ' AS ');
if ($pos) {
$alias = $this->replaceName(substr($quoted, $pos + 4));
$quoted = substr($quoted, 0, $pos) . " AS $alias";
}
return $quoted;
}
/**
* Quotes name
*
* @param string $name
* @return string
*/
protected function replaceName($name)
{
$name = trim($name);
if ($name == '*') {
return $name;
}
return '`' . $name . '`';
}
/**
* Quotes
*
* @param string $text
* @return string|array
*/
protected function replaceNamesIn($text)
{
$is_string_literal = strpos($text, "'") !== false
|| strpos($text, '"') !== false;
if ($is_string_literal) {
return $text;
}
$word = '[a-z_][a-z0-9_]+';
$find = "/(\\b)($word)\\.($word)(\\b)/i";
$repl = '$1`$2`.`$3`$4';
$text = preg_replace($find, $repl, $text);
return $text;
}
// ---------- insert --------------
/**
* 设置 `table.column` 与 last-insert-id 的映射
*
* @param array $last_insert_id_names
*/
public function setLastInsertIdNames(array $last_insert_id_names)
{
$this->last_insert_id_names = $last_insert_id_names;
}
/**
* insert into.
*
* @param string $table
* @return self
*/
public function into($table)
{
$this->table = $this->quoteName($table);
return $this;
}
/**
* 生成 INSERT 语句
*
* @return string
*/
protected function buildINSERT()
{
return 'INSERT'
. $this->buildFlags()
. $this->buildInto()
. $this->buildValuesForInsert()
. $this->buildReturning();
}
/**
* 生成 INTO 语句
*
* @return string
*/
protected function buildInto()
{
return " INTO " . $this->table;
}
/**
* PDO::lastInsertId()
*
* @param string $col
* @return mixed
*/
public function getLastInsertIdName($col)
{
$key = str_replace('`', '', $this->table) . '.' . $col;
if (isset($this->last_insert_id_names[$key])) {
return $this->last_insert_id_names[$key];
}
return null;
}
/**
* 设置一列,如果有第二各参数,则把第二个参数绑定在占位符上
*
* @param string $col
* @return self
*/
public function col($col)
{
return call_user_func_array(array($this, 'addCol'), func_get_args());
}
/**
* 设置多列
*
* @param array $cols
* @return self
*/
public function cols(array $cols)
{
if ($this->type == 'SELECT') {
foreach ($cols as $key => $val) {
$this->addColSELECT($key, $val);
}
return $this;
}
return $this->addCols($cols);
}
/**
* 直接设置列的值
*
* @param string $col
* @param string $value
* @return self
*/
public function set($col, $value)
{
return $this->setCol($col, $value);
}
/**
* 为 INSERT 语句绑定值
*
* @return string
*/
protected function buildValuesForInsert()
{
return ' (' . $this->indentCsv(array_keys($this->col_values)) . ') VALUES (' .
$this->indentCsv(array_values($this->col_values)) . ')';
}
// ------update-------
/**
* 更新哪个表
*
* @param string $table
* @return self
*/
public function table($table)
{
$this->table = $this->quoteName($table);
return $this;
}
/**
* 生成完整 SQL 语句
*
* @return string
* @throws Exception
*/
protected function build()
{
switch ($this->type) {
case 'DELETE':
return $this->buildDELETE();
case 'INSERT':
return $this->buildINSERT();
case 'UPDATE':
return $this->buildUPDATE();
case 'SELECT':
return $this->buildSELECT();
}
throw new Exception("type empty");
}
/**
* 生成更新的 SQL 语句
*/
protected function buildUPDATE()
{
return 'UPDATE'
. $this->buildFlags()
. $this->buildTable()
. $this->buildValuesForUpdate()
. $this->buildWhere()
. $this->buildOrderBy()
. $this->buildLimit()
. $this->buildReturning();
}
/**
* 哪个表
*
* @return string
*/
protected function buildTable()
{
return " {$this->table}";
}
/**
* 为更新语句绑定值
*
* @return string
*/
protected function buildValuesForUpdate()
{
$values = array();
foreach ($this->col_values as $col => $value) {
$values[] = "{$col} = {$value}";
}
return ' SET' . $this->indentCsv($values);
}
// ----------Dml---------------
/**
* 获取绑定的值
*
* @return array
*/
public function getBindValuesCOMMON()
{
$bind_values = $this->bind_values;
$i = 1;
foreach ($this->bind_where as $val) {
$bind_values[$i] = $val;
$i++;
}
return $bind_values;
}
/**
* 设置列
*
* @param string $col
* @return self
*/
protected function addCol($col)
{
$key = $this->quoteName($col);
$this->col_values[$key] = ":$col";
$args = func_get_args();
if (count($args) > 1) {
$this->bindValue($col, $args[1]);
}
return $this;
}
/**
* 设置多个列
*
* @param array $cols
* @return self
*/
protected function addCols(array $cols)
{
foreach ($cols as $key => $val) {
if (is_int($key)) {
$this->addCol($val);
} else {
$this->addCol($key, $val);
}
}
return $this;
}
/**
* 设置单列的值
*
* @param string $col .
* @param string $value
* @return self
*/
protected function setCol($col, $value)
{
if ($value === null) {
$value = 'NULL';
}
$key = $this->quoteName($col);
$value = $this->quoteNamesIn($value);
$this->col_values[$key] = $value;
return $this;
}
/**
* 增加返回的列
*
* @param array $cols
* @return self
*
*/
protected function addReturning(array $cols)
{
foreach ($cols as $col) {
$this->returning[] = $this->quoteNamesIn($col);
}
return $this;
}
/**
* 生成 RETURNING 语句
*
* @return string
*/
protected function buildReturning()
{
if (!$this->returning) {
return '';
}
return ' RETURNING' . $this->indentCsv($this->returning);
}
/**
* 构造函数
*
* @param string $host
* @param int $port
* @param string $user
* @param string $password
* @param string $db_name
* @param string $charset
*/
public function __construct($host, $port, $user, $password, $db_name, $charset = 'utf8')
{
$this->settings = array(
'host' => $host,
'port' => $port,
'user' => $user,
'password' => $password,
'dbname' => $db_name,
'charset' => $charset,
);
$this->connect();
}
/**
* 创建 PDO 实例
*/
protected function connect()
{
$dsn = 'mysql:dbname=' . $this->settings["dbname"] . ';host=' .
$this->settings["host"] . ';port=' . $this->settings['port'];
$this->pdo = new PDO($dsn, $this->settings["user"], $this->settings["password"],
array(
PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES ' . (!empty($this->settings['charset']) ?
$this->settings['charset'] : 'utf8')
));
$this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$this->pdo->setAttribute(PDO::ATTR_STRINGIFY_FETCHES, false);
$this->pdo->setAttribute(PDO::ATTR_EMULATE_PREPARES, false);
}
/**
* 关闭连接
*/
public function closeConnection()
{
$this->pdo = null;
}
/**
* 执行
*
* @param string $query
* @param string $parameters
* @throws PDOException
*/
protected function execute($query, $parameters = "")
{
try {
$this->sQuery = @$this->pdo->prepare($query);
$this->bindMore($parameters);
if (!empty($this->parameters)) {
foreach ($this->parameters as $param) {
$parameters = explode("\x7F", $param);
$this->sQuery->bindParam($parameters[0], $parameters[1]);
}
}
$this->success = $this->sQuery->execute();
} catch (PDOException $e) {
// 服务端断开时重连一次
if ($e->errorInfo[1] == 2006 || $e->errorInfo[1] == 2013) {
$this->closeConnection();
$this->connect();
try {
$this->sQuery = $this->pdo->prepare($query);
$this->bindMore($parameters);
if (!empty($this->parameters)) {
foreach ($this->parameters as $param) {
$parameters = explode("\x7F", $param);
$this->sQuery->bindParam($parameters[0], $parameters[1]);
}
}
$this->success = $this->sQuery->execute();
} catch (PDOException $ex) {
$this->rollBackTrans();
throw $ex;
}
} else {
$this->rollBackTrans();
$msg = $e->getMessage();
$err_msg = "SQL:".$this->lastSQL()." ".$msg;
$exception = new \PDOException($err_msg, (int)$e->getCode());
throw $exception;
}
}
$this->parameters = array();
}
/**
* 绑定
*
* @param string $para
* @param string $value
*/
public function bind($para, $value)
{
if (is_string($para)) {
$this->parameters[sizeof($this->parameters)] = ":" . $para . "\x7F" . $value;
} else {
$this->parameters[sizeof($this->parameters)] = $para . "\x7F" . $value;
}
}
/**
* 绑定多个
*
* @param array $parray
*/
public function bindMore($parray)
{
if (empty($this->parameters) && is_array($parray)) {
$columns = array_keys($parray);
foreach ($columns as $i => &$column) {
$this->bind($column, $parray[$column]);
}
}
}
/**
* 执行 SQL
*
* @param string $query
* @param array $params
* @param int $fetchmode
* @return mixed
*/
public function query($query = '', $params = null, $fetchmode = PDO::FETCH_ASSOC)
{
$query = trim($query);
if (empty($query)) {
$query = $this->build();
if (!$params) {
$params = $this->getBindValues();
}
}
$this->resetAll();
$this->lastSql = $query;
$this->execute($query, $params);
$rawStatement = explode(" ", $query);
$statement = strtolower(trim($rawStatement[0]));
if ($statement === 'select' || $statement === 'show') {
return $this->sQuery->fetchAll($fetchmode);
} elseif ($statement === 'update' || $statement === 'delete') {
return $this->sQuery->rowCount();
} elseif ($statement === 'insert') {
if ($this->sQuery->rowCount() > 0) {
return $this->lastInsertId();
}
} else {
return null;
}
return null;
}
/**
* 返回一列
*
* @param string $query
* @param array $params
* @return array
*/
public function column($query = '', $params = null)
{
$query = trim($query);
if (empty($query)) {
$query = $this->build();
if (!$params) {
$params = $this->getBindValues();
}
}
$this->resetAll();
$this->lastSql = $query;
$this->execute($query, $params);
$columns = $this->sQuery->fetchAll(PDO::FETCH_NUM);
$column = null;
foreach ($columns as $cells) {
$column[] = $cells[0];
}
return $column;
}
/**
* 返回一行
*
* @param string $query
* @param array $params
* @param int $fetchmode
* @return array
*/
public function row($query = '', $params = null, $fetchmode = PDO::FETCH_ASSOC)
{
$query = trim($query);
if (empty($query)) {
$query = $this->build();
if (!$params) {
$params = $this->getBindValues();
}
}
$this->resetAll();
$this->lastSql = $query;
$this->execute($query, $params);
return $this->sQuery->fetch($fetchmode);
}
/**
* 返回单个值
*
* @param string $query
* @param array $params
* @return string
*/
public function single($query = '', $params = null)
{
$query = trim($query);
if (empty($query)) {
$query = $this->build();
if (!$params) {
$params = $this->getBindValues();
}
}
$this->resetAll();
$this->lastSql = $query;
$this->execute($query, $params);
return $this->sQuery->fetchColumn();
}
/**
* 返回 lastInsertId
*
* @return string
*/
public function lastInsertId()
{
return $this->pdo->lastInsertId();
}
/**
* 返回最后一条执行的 sql
*
* @return string
*/
public function lastSQL()
{
return $this->lastSql;
}
/**
* 开始事务
*/
public function beginTrans()
{
try {
$this->pdo->beginTransaction();
} catch (PDOException $e) {
// 服务端断开时重连一次
if ($e->errorInfo[1] == 2006 || $e->errorInfo[1] == 2013) {
$this->pdo->beginTransaction();
} else {
throw $e;
}
}
}
/**
* 提交事务
*/
public function commitTrans()
{
$this->pdo->commit();
}
/**
* 事务回滚
*/
public function rollBackTrans()
{
if ($this->pdo->inTransaction()) {
$this->pdo->rollBack();
}
}
}
@@ -0,0 +1,1361 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<[email protected]>
* @copyright walkor<[email protected]>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace GatewayWorker\Lib;
use Exception;
use GatewayWorker\Protocols\GatewayProtocol;
use Workerman\Connection\TcpConnection;
/**
* 数据发送相关
*/
class Gateway
{
/**
* gateway 实例
*
* @var object
*/
protected static $businessWorker = null;
/**
* 注册中心地址
*
* @var string|array
*/
public static $registerAddress = '127.0.0.1:1236';
/**
* 秘钥
* @var string
*/
public static $secretKey = '';
/**
* 链接超时时间
* @var int
*/
public static $connectTimeout = 3;
/**
* 与Gateway是否是长链接
* @var bool
*/
public static $persistentConnection = false;
/**
* 向所有客户端连接(或者 client_id_array 指定的客户端连接)广播消息
*
* @param string $message 向客户端发送的消息
* @param array $client_id_array 客户端 id 数组
* @param array $exclude_client_id 不给这些client_id发
* @param bool $raw 是否发送原始数据(即不调用gateway的协议的encode方法)
* @return void
* @throws Exception
*/
public static function sendToAll($message, $client_id_array = null, $exclude_client_id = null, $raw = false)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_SEND_TO_ALL;
$gateway_data['body'] = $message;
if ($raw) {
$gateway_data['flag'] |= GatewayProtocol::FLAG_NOT_CALL_ENCODE;
}
if ($exclude_client_id) {
if (!is_array($exclude_client_id)) {
$exclude_client_id = array($exclude_client_id);
}
if ($client_id_array) {
$exclude_client_id = array_flip($exclude_client_id);
}
}
if ($client_id_array) {
if (!is_array($client_id_array)) {
echo new \Exception('bad $client_id_array:'.var_export($client_id_array, true));
return;
}
$data_array = array();
foreach ($client_id_array as $client_id) {
if (isset($exclude_client_id[$client_id])) {
continue;
}
$address = Context::clientIdToAddress($client_id);
if ($address) {
$key = long2ip($address['local_ip']) . ":{$address['local_port']}";
$data_array[$key][$address['connection_id']] = $address['connection_id'];
}
}
foreach ($data_array as $addr => $connection_id_list) {
$the_gateway_data = $gateway_data;
$the_gateway_data['ext_data'] = json_encode(array('connections' => $connection_id_list));
static::sendToGateway($addr, $the_gateway_data);
}
return;
} elseif (empty($client_id_array) && is_array($client_id_array)) {
return;
}
if (!$exclude_client_id) {
return static::sendToAllGateway($gateway_data);
}
$address_connection_array = static::clientIdArrayToAddressArray($exclude_client_id);
// 如果有businessWorker实例,说明运行在workerman环境中,通过businessWorker中的长连接发送数据
if (static::$businessWorker) {
foreach (static::$businessWorker->gatewayConnections as $address => $gateway_connection) {
$gateway_data['ext_data'] = isset($address_connection_array[$address]) ?
json_encode(array('exclude'=> $address_connection_array[$address])) : '';
/** @var TcpConnection $gateway_connection */
$gateway_connection->send($gateway_data);
}
} // 运行在其它环境中,通过注册中心得到gateway地址
else {
$all_addresses = static::getAllGatewayAddressesFromRegister();
foreach ($all_addresses as $address) {
$gateway_data['ext_data'] = isset($address_connection_array[$address]) ?
json_encode(array('exclude'=> $address_connection_array[$address])) : '';
static::sendToGateway($address, $gateway_data);
}
}
}
/**
* 向某个client_id对应的连接发消息
*
* @param int $client_id
* @param string $message
* @return void
*/
public static function sendToClient($client_id, $message)
{
return static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_SEND_TO_ONE, $message);
}
/**
* 向当前客户端连接发送消息
*
* @param string $message
* @return bool
*/
public static function sendToCurrentClient($message)
{
return static::sendCmdAndMessageToClient(null, GatewayProtocol::CMD_SEND_TO_ONE, $message);
}
/**
* 判断某个uid是否在线
*
* @param string $uid
* @return int 0|1
*/
public static function isUidOnline($uid)
{
return (int)static::getClientIdByUid($uid);
}
/**
* 判断client_id对应的连接是否在线
*
* @param int $client_id
* @return int 0|1
*/
public static function isOnline($client_id)
{
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return 0;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
if (isset(static::$businessWorker)) {
if (!isset(static::$businessWorker->gatewayConnections[$address])) {
return 0;
}
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_IS_ONLINE;
$gateway_data['connection_id'] = $address_data['connection_id'];
return (int)static::sendAndRecv($address, $gateway_data);
}
/**
* 获取所有在线用户的sessionclient_id为 key(弃用,请用getAllClientSessions代替)
*
* @param string $group
* @return array
*/
public static function getAllClientInfo($group = '')
{
echo "Warning: Gateway::getAllClientInfo is deprecated and will be removed in a future, please use Gateway::getAllClientSessions instead.";
return static::getAllClientSessions($group);
}
/**
* 获取所有在线client_id的sessionclient_id为 key
*
* @param string $group
* @return array
*/
public static function getAllClientSessions($group = '')
{
$gateway_data = GatewayProtocol::$empty;
if (!$group) {
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_ALL_CLIENT_SESSIONS;
} else {
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_CLIENT_SESSIONS_BY_GROUP;
$gateway_data['ext_data'] = $group;
}
$status_data = array();
$all_buffer_array = static::getBufferFromAllGateway($gateway_data);
foreach ($all_buffer_array as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $data) {
if ($data) {
foreach ($data as $connection_id => $session_buffer) {
$client_id = Context::addressToClientId($local_ip, $local_port, $connection_id);
if ($client_id === Context::$client_id) {
$status_data[$client_id] = (array)$_SESSION;
} else {
$status_data[$client_id] = $session_buffer ? Context::sessionDecode($session_buffer) : array();
}
}
}
}
}
return $status_data;
}
/**
* 获取某个组的连接信息(弃用,请用getClientSessionsByGroup代替)
*
* @param string $group
* @return array
*/
public static function getClientInfoByGroup($group)
{
echo "Warning: Gateway::getClientInfoByGroup is deprecated and will be removed in a future, please use Gateway::getClientSessionsByGroup instead.";
return static::getAllClientSessions($group);
}
/**
* 获取某个组的所有client_id的session信息
*
* @param string $group
*
* @return array
*/
public static function getClientSessionsByGroup($group)
{
if (static::isValidGroupId($group)) {
return static::getAllClientSessions($group);
}
return array();
}
/**
* 获取所有在线client_id数
*
* @return int
*/
public static function getAllClientIdCount()
{
return static::getClientCountByGroup();
}
/**
* 获取所有在线client_id数(getAllClientIdCount的别名)
*
* @return int
*/
public static function getAllClientCount()
{
return static::getAllClientIdCount();
}
/**
* 获取某个组的在线client_id数
*
* @param string $group
* @return int
*/
public static function getClientIdCountByGroup($group = '')
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_CLIENT_COUNT_BY_GROUP;
$gateway_data['ext_data'] = $group;
$total_count = 0;
$all_buffer_array = static::getBufferFromAllGateway($gateway_data);
foreach ($all_buffer_array as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $count) {
if ($count) {
$total_count += $count;
}
}
}
return $total_count;
}
/**
* getClientIdCountByGroup 函数的别名
*
* @param string $group
* @return int
*/
public static function getClientCountByGroup($group = '')
{
return static::getClientIdCountByGroup($group);
}
/**
* 获取某个群组在线client_id列表
*
* @param string $group
* @return array
*/
public static function getClientIdListByGroup($group)
{
if (!static::isValidGroupId($group)) {
return array();
}
$data = static::select(array('uid'), array('groups' => is_array($group) ? $group : array($group)));
$client_id_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
$client_id = Context::addressToClientId($local_ip, $local_port, $connection_id);
$client_id_map[$client_id] = $client_id;
}
}
}
return $client_id_map;
}
/**
* 获取集群所有在线client_id列表
*
* @return array
*/
public static function getAllClientIdList()
{
return static::formatClientIdFromGatewayBuffer(static::select(array('uid')));
}
/**
* 格式化client_id
*
* @param $data
* @return array
*/
protected static function formatClientIdFromGatewayBuffer($data)
{
$client_id_list = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
$client_id = Context::addressToClientId($local_ip, $local_port, $connection_id);
$client_id_list[$client_id] = $client_id;
}
}
}
return $client_id_list;
}
/**
* 获取与 uid 绑定的 client_id 列表
*
* @param string $uid
* @return array
*/
public static function getClientIdByUid($uid)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_CLIENT_ID_BY_UID;
$gateway_data['ext_data'] = $uid;
$client_list = array();
$all_buffer_array = static::getBufferFromAllGateway($gateway_data);
foreach ($all_buffer_array as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $connection_id_array) {
if ($connection_id_array) {
foreach ($connection_id_array as $connection_id) {
$client_list[] = Context::addressToClientId($local_ip, $local_port, $connection_id);
}
}
}
}
return $client_list;
}
/**
* 获取某个群组在线uid列表
*
* @param string $group
* @return array
*/
public static function getUidListByGroup($group)
{
if (!static::isValidGroupId($group)) {
return array();
}
$group = is_array($group) ? $group : array($group);
$data = static::select(array('uid'), array('groups' => $group));
$uid_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
if (!empty($info['uid'])) {
$uid_map[$info['uid']] = $info['uid'];
}
}
}
}
return $uid_map;
}
/**
* 获取某个群组在线uid数
*
* @param string $group
* @return int
*/
public static function getUidCountByGroup($group)
{
if (static::isValidGroupId($group)) {
return count(static::getUidListByGroup($group));
}
return 0;
}
/**
* 获取全局在线uid列表
*
* @return array
*/
public static function getAllUidList()
{
$data = static::select(array('uid'));
$uid_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
if (!empty($info['uid'])) {
$uid_map[$info['uid']] = $info['uid'];
}
}
}
}
return $uid_map;
}
/**
* 获取全局在线uid数
* @return int
*/
public static function getAllUidCount()
{
return count(static::getAllUidList());
}
/**
* 通过client_id获取uid
*
* @param $client_id
* @return mixed
*/
public static function getUidByClientId($client_id)
{
$data = static::select(array('uid'), array('client_id'=>array($client_id)));
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $info) {
return $info['uid'];
}
}
}
}
/**
* 获取所有在线的群组id
*
* @return array
*/
public static function getAllGroupIdList()
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_GROUP_ID_LIST;
$group_id_list = array();
$all_buffer_array = static::getBufferFromAllGateway($gateway_data);
foreach ($all_buffer_array as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $group_id_array) {
if (is_array($group_id_array)) {
foreach ($group_id_array as $group_id) {
if (!isset($group_id_list[$group_id])) {
$group_id_list[$group_id] = $group_id;
}
}
}
}
}
return $group_id_list;
}
/**
* 获取所有在线分组的uid数量,也就是每个分组的在线用户数
*
* @return array
*/
public static function getAllGroupUidCount()
{
$group_uid_map = static::getAllGroupUidList();
$group_uid_count_map = array();
foreach ($group_uid_map as $group_id => $uid_list) {
$group_uid_count_map[$group_id] = count($uid_list);
}
return $group_uid_count_map;
}
/**
* 获取所有分组uid在线列表
*
* @return array
*/
public static function getAllGroupUidList()
{
$data = static::select(array('uid','groups'));
$group_uid_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
if (empty($info['uid']) || empty($info['groups'])) {
break;
}
$uid = $info['uid'];
foreach ($info['groups'] as $group_id) {
if(!isset($group_uid_map[$group_id])) {
$group_uid_map[$group_id] = array();
}
$group_uid_map[$group_id][$uid] = $uid;
}
}
}
}
return $group_uid_map;
}
/**
* 获取所有群组在线client_id列表
*
* @return array
*/
public static function getAllGroupClientIdList()
{
$data = static::select(array('groups'));
$group_client_id_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
if (empty($info['groups'])) {
break;
}
$client_id = Context::addressToClientId($local_ip, $local_port, $connection_id);
foreach ($info['groups'] as $group_id) {
if(!isset($group_client_id_map[$group_id])) {
$group_client_id_map[$group_id] = array();
}
$group_client_id_map[$group_id][$client_id] = $client_id;
}
}
}
}
return $group_client_id_map;
}
/**
* 获取所有群组在线client_id数量,也就是获取每个群组在线连接数
*
* @return array
*/
public static function getAllGroupClientIdCount()
{
$group_client_map = static::getAllGroupClientIdList();
$group_client_count_map = array();
foreach ($group_client_map as $group_id => $client_id_list) {
$group_client_count_map[$group_id] = count($client_id_list);
}
return $group_client_count_map;
}
/**
* 根据条件到gateway搜索数据
*
* @param array $fields
* @param array $where
* @return array
*/
protected static function select($fields = array('session','uid','groups'), $where = array())
{
$t = microtime(true);
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_SELECT;
$gateway_data['ext_data'] = array('fields' => $fields, 'where' => $where);
$gateway_data_list = array();
// 有client_id,能计算出需要和哪些gateway通讯,只和必要的gateway通讯能降低系统负载
if (isset($where['client_id'])) {
$client_id_list = $where['client_id'];
unset($gateway_data['ext_data']['where']['client_id']);
$gateway_data['ext_data']['where']['connection_id'] = array();
foreach ($client_id_list as $client_id) {
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
continue;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
if (!isset($gateway_data_list[$address])) {
$gateway_data_list[$address] = $gateway_data;
}
$gateway_data_list[$address]['ext_data']['where']['connection_id'][$address_data['connection_id']] = $address_data['connection_id'];
}
foreach ($gateway_data_list as $address => $item) {
$gateway_data_list[$address]['ext_data'] = json_encode($item['ext_data']);
}
// 有其它条件,则还是需要向所有gateway发送
if (count($where) !== 1) {
$gateway_data['ext_data'] = json_encode($gateway_data['ext_data']);
foreach (static::getAllGatewayAddress() as $address) {
if (!isset($gateway_data_list[$address])) {
$gateway_data_list[$address] = $gateway_data;
}
}
}
$data = static::getBufferFromSomeGateway($gateway_data_list);
} else {
$gateway_data['ext_data'] = json_encode($gateway_data['ext_data']);
$data = static::getBufferFromAllGateway($gateway_data);
}
return $data;
}
/**
* 生成验证包,用于验证此客户端的合法性
*
* @return string
*/
protected static function generateAuthBuffer()
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GATEWAY_CLIENT_CONNECT;
$gateway_data['body'] = json_encode(array(
'secret_key' => static::$secretKey,
));
return GatewayProtocol::encode($gateway_data);
}
/**
* 批量向某些gateway发包,并得到返回数组
*
* @param array $gateway_data_array
* @return array
* @throws Exception
*/
protected static function getBufferFromSomeGateway($gateway_data_array)
{
$gateway_buffer_array = array();
$auth_buffer = static::$secretKey ? static::generateAuthBuffer() : '';
foreach ($gateway_data_array as $address => $gateway_data) {
if ($auth_buffer) {
$gateway_buffer_array[$address] = $auth_buffer.GatewayProtocol::encode($gateway_data);
} else {
$gateway_buffer_array[$address] = GatewayProtocol::encode($gateway_data);
}
}
return static::getBufferFromGateway($gateway_buffer_array);
}
/**
* 批量向所有 gateway 发包,并得到返回数组
*
* @param string $gateway_data
* @return array
* @throws Exception
*/
protected static function getBufferFromAllGateway($gateway_data)
{
$addresses = static::getAllGatewayAddress();
$gateway_buffer_array = array();
$gateway_buffer = GatewayProtocol::encode($gateway_data);
$gateway_buffer = static::$secretKey ? static::generateAuthBuffer() . $gateway_buffer : $gateway_buffer;
foreach ($addresses as $address) {
$gateway_buffer_array[$address] = $gateway_buffer;
}
return static::getBufferFromGateway($gateway_buffer_array);
}
/**
* 获取所有gateway内部通讯地址
*
* @return array
* @throws Exception
*/
protected static function getAllGatewayAddress()
{
if (isset(static::$businessWorker)) {
$addresses = static::$businessWorker->getAllGatewayAddresses();
if (empty($addresses)) {
throw new Exception('businessWorker::getAllGatewayAddresses return empty');
}
} else {
$addresses = static::getAllGatewayAddressesFromRegister();
if (empty($addresses)) {
return array();
}
}
return $addresses;
}
/**
* 批量向gateway发送并获取数据
* @param $gateway_buffer_array
* @return array
*/
protected static function getBufferFromGateway($gateway_buffer_array)
{
$client_array = $status_data = $client_address_map = $receive_buffer_array = $recv_length_array = array();
// 批量向所有gateway进程发送请求数据
foreach ($gateway_buffer_array as $address => $gateway_buffer) {
$client = stream_socket_client("tcp://$address", $errno, $errmsg, static::$connectTimeout);
if ($client && strlen($gateway_buffer) === stream_socket_sendto($client, $gateway_buffer)) {
$socket_id = (int)$client;
$client_array[$socket_id] = $client;
$client_address_map[$socket_id] = explode(':', $address);
$receive_buffer_array[$socket_id] = '';
}
}
// 超时5秒
$timeout = 5;
$time_start = microtime(true);
// 批量接收请求
while (count($client_array) > 0) {
$write = $except = array();
$read = $client_array;
if (@stream_select($read, $write, $except, $timeout)) {
foreach ($read as $client) {
$socket_id = (int)$client;
$buffer = stream_socket_recvfrom($client, 65535);
if ($buffer !== '' && $buffer !== false) {
$receive_buffer_array[$socket_id] .= $buffer;
$receive_length = strlen($receive_buffer_array[$socket_id]);
if (empty($recv_length_array[$socket_id]) && $receive_length >= 4) {
$recv_length_array[$socket_id] = current(unpack('N', $receive_buffer_array[$socket_id]));
}
if (!empty($recv_length_array[$socket_id]) && $receive_length >= $recv_length_array[$socket_id] + 4) {
unset($client_array[$socket_id]);
}
} elseif (feof($client)) {
unset($client_array[$socket_id]);
}
}
}
if (microtime(true) - $time_start > $timeout) {
break;
}
}
$format_buffer_array = array();
foreach ($receive_buffer_array as $socket_id => $buffer) {
$local_ip = ip2long($client_address_map[$socket_id][0]);
$local_port = $client_address_map[$socket_id][1];
$format_buffer_array[$local_ip][$local_port] = unserialize(substr($buffer, 4));
}
return $format_buffer_array;
}
/**
* 踢掉某个客户端,并以$message通知被踢掉客户端
*
* @param int $client_id
* @param string $message
* @return void
*/
public static function closeClient($client_id, $message = null)
{
if ($client_id === Context::$client_id) {
return static::closeCurrentClient($message);
} // 不是发给当前用户则使用存储中的地址
else {
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return false;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
return static::kickAddress($address, $address_data['connection_id'], $message);
}
}
/**
* 踢掉当前客户端,并以$message通知被踢掉客户端
*
* @param string $message
* @return bool
* @throws Exception
*/
public static function closeCurrentClient($message = null)
{
if (!Context::$connection_id) {
throw new Exception('closeCurrentClient can not be called in async context');
}
$address = long2ip(Context::$local_ip) . ':' . Context::$local_port;
return static::kickAddress($address, Context::$connection_id, $message);
}
/**
* 踢掉某个客户端并直接立即销毁相关连接
*
* @param int $client_id
* @return bool
*/
public static function destoryClient($client_id)
{
if ($client_id === Context::$client_id) {
return static::destoryCurrentClient();
} // 不是发给当前用户则使用存储中的地址
else {
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return false;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
return static::destroyAddress($address, $address_data['connection_id']);
}
}
/**
* 踢掉当前客户端并直接立即销毁相关连接
*
* @return bool
* @throws Exception
*/
public static function destoryCurrentClient()
{
if (!Context::$connection_id) {
throw new Exception('destoryCurrentClient can not be called in async context');
}
$address = long2ip(Context::$local_ip) . ':' . Context::$local_port;
return static::destroyAddress($address, Context::$connection_id);
}
/**
* 将 client_id 与 uid 绑定
*
* @param int $client_id
* @param int|string $uid
* @return void
*/
public static function bindUid($client_id, $uid)
{
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_BIND_UID, '', $uid);
}
/**
* 将 client_id 与 uid 解除绑定
*
* @param int $client_id
* @param int|string $uid
* @return void
*/
public static function unbindUid($client_id, $uid)
{
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_UNBIND_UID, '', $uid);
}
/**
* 将 client_id 加入组
*
* @param int $client_id
* @param int|string $group
* @return void
*/
public static function joinGroup($client_id, $group)
{
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_JOIN_GROUP, '', $group);
}
/**
* 将 client_id 离开组
*
* @param int $client_id
* @param int|string $group
*
* @return void
*/
public static function leaveGroup($client_id, $group)
{
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_LEAVE_GROUP, '', $group);
}
/**
* 取消分组
*
* @param int|string $group
*
* @return void
*/
public static function ungroup($group)
{
if (!static::isValidGroupId($group)) {
return false;
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_UNGROUP;
$gateway_data['ext_data'] = $group;
return static::sendToAllGateway($gateway_data);
}
/**
* 向所有 uid 发送
*
* @param int|string|array $uid
* @param string $message
*
* @return void
*/
public static function sendToUid($uid, $message)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_SEND_TO_UID;
$gateway_data['body'] = $message;
if (!is_array($uid)) {
$uid = array($uid);
}
$gateway_data['ext_data'] = json_encode($uid);
static::sendToAllGateway($gateway_data);
}
/**
* 向 group 发送
*
* @param int|string|array $group 组(不允许是 0 '0' false null array()等为空的值)
* @param string $message 消息
* @param array $exclude_client_id 不给这些client_id发
* @param bool $raw 发送原始数据(即不调用gateway的协议的encode方法)
*
* @return void
*/
public static function sendToGroup($group, $message, $exclude_client_id = null, $raw = false)
{
if (!static::isValidGroupId($group)) {
return false;
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_SEND_TO_GROUP;
$gateway_data['body'] = $message;
if ($raw) {
$gateway_data['flag'] |= GatewayProtocol::FLAG_NOT_CALL_ENCODE;
}
if (!is_array($group)) {
$group = array($group);
}
// 分组发送,没有排除的client_id,直接发送
$default_ext_data_buffer = json_encode(array('group'=> $group, 'exclude'=> null));
if (empty($exclude_client_id)) {
$gateway_data['ext_data'] = $default_ext_data_buffer;
return static::sendToAllGateway($gateway_data);
}
// 分组发送,有排除的client_id,需要将client_id转换成对应gateway进程内的connectionId
if (!is_array($exclude_client_id)) {
$exclude_client_id = array($exclude_client_id);
}
$address_connection_array = static::clientIdArrayToAddressArray($exclude_client_id);
// 如果有businessWorker实例,说明运行在workerman环境中,通过businessWorker中的长连接发送数据
if (static::$businessWorker) {
foreach (static::$businessWorker->gatewayConnections as $address => $gateway_connection) {
$gateway_data['ext_data'] = isset($address_connection_array[$address]) ?
json_encode(array('group'=> $group, 'exclude'=> $address_connection_array[$address])) :
$default_ext_data_buffer;
/** @var TcpConnection $gateway_connection */
$gateway_connection->send($gateway_data);
}
} // 运行在其它环境中,通过注册中心得到gateway地址
else {
$addresses = static::getAllGatewayAddressesFromRegister();
foreach ($addresses as $address) {
$gateway_data['ext_data'] = isset($address_connection_array[$address]) ?
json_encode(array('group'=> $group, 'exclude'=> $address_connection_array[$address])) :
$default_ext_data_buffer;
static::sendToGateway($address, $gateway_data);
}
}
}
/**
* 更新 session,框架自动调用,开发者不要调用
*
* @param int $client_id
* @param string $session_str
* @return bool
*/
public static function setSocketSession($client_id, $session_str)
{
return static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_SET_SESSION, '', $session_str);
}
/**
* 设置 session,原session值会被覆盖
*
* @param int $client_id
* @param array $session
*
* @return void
*/
public static function setSession($client_id, array $session)
{
if (Context::$client_id === $client_id) {
$_SESSION = $session;
Context::$old_session = $_SESSION;
}
static::setSocketSession($client_id, Context::sessionEncode($session));
}
/**
* 更新 session,实际上是与老的session合并
*
* @param int $client_id
* @param array $session
*
* @return void
*/
public static function updateSession($client_id, array $session)
{
if (Context::$client_id === $client_id) {
$_SESSION = array_replace_recursive((array)$_SESSION, $session);
Context::$old_session = $_SESSION;
}
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_UPDATE_SESSION, '', Context::sessionEncode($session));
}
/**
* 获取某个client_id的session
*
* @param int $client_id
* @return mixed false表示出错、null表示用户不存在、array表示具体的session信息
*/
public static function getSession($client_id)
{
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return false;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
if (isset(static::$businessWorker)) {
if (!isset(static::$businessWorker->gatewayConnections[$address])) {
return null;
}
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_SESSION_BY_CLIENT_ID;
$gateway_data['connection_id'] = $address_data['connection_id'];
return static::sendAndRecv($address, $gateway_data);
}
/**
* 向某个用户网关发送命令和消息
*
* @param int $client_id
* @param int $cmd
* @param string $message
* @param string $ext_data
* @return boolean
*/
protected static function sendCmdAndMessageToClient($client_id, $cmd, $message, $ext_data = '')
{
// 如果是发给当前用户则直接获取上下文中的地址
if ($client_id === Context::$client_id || $client_id === null) {
$address = long2ip(Context::$local_ip) . ':' . Context::$local_port;
$connection_id = Context::$connection_id;
} else {
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return false;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
$connection_id = $address_data['connection_id'];
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = $cmd;
$gateway_data['connection_id'] = $connection_id;
$gateway_data['body'] = $message;
if (!empty($ext_data)) {
$gateway_data['ext_data'] = $ext_data;
}
return static::sendToGateway($address, $gateway_data);
}
/**
* 发送数据并返回
*
* @param int $address
* @param mixed $data
* @return bool
* @throws Exception
*/
protected static function sendAndRecv($address, $data)
{
$buffer = GatewayProtocol::encode($data);
$buffer = static::$secretKey ? static::generateAuthBuffer() . $buffer : $buffer;
$client = stream_socket_client("tcp://$address", $errno, $errmsg, static::$connectTimeout);
if (!$client) {
throw new Exception("can not connect to tcp://$address $errmsg");
}
if (strlen($buffer) === stream_socket_sendto($client, $buffer)) {
$timeout = 5;
// 阻塞读
stream_set_blocking($client, 1);
// 1秒超时
stream_set_timeout($client, 1);
$all_buffer = '';
$time_start = microtime(true);
$pack_len = 0;
while (1) {
$buf = stream_socket_recvfrom($client, 655350);
if ($buf !== '' && $buf !== false) {
$all_buffer .= $buf;
} else {
if (feof($client)) {
throw new Exception("connection close tcp://$address");
} elseif (microtime(true) - $time_start > $timeout) {
break;
}
continue;
}
$recv_len = strlen($all_buffer);
if (!$pack_len && $recv_len >= 4) {
$pack_len= current(unpack('N', $all_buffer));
}
// 回复的数据都是以\n结尾
if (($pack_len && $recv_len >= $pack_len + 4) || microtime(true) - $time_start > $timeout) {
break;
}
}
// 返回结果
return unserialize(substr($all_buffer, 4));
} else {
throw new Exception("sendAndRecv($address, \$bufer) fail ! Can not send data!", 502);
}
}
/**
* 发送数据到网关
*
* @param string $address
* @param array $gateway_data
* @return bool
*/
protected static function sendToGateway($address, $gateway_data)
{
return static::sendBufferToGateway($address, GatewayProtocol::encode($gateway_data));
}
/**
* 发送buffer数据到网关
* @param string $address
* @param string $gateway_buffer
* @return bool
*/
protected static function sendBufferToGateway($address, $gateway_buffer)
{
// 有$businessWorker说明是workerman环境,使用$businessWorker发送数据
if (static::$businessWorker) {
if (!isset(static::$businessWorker->gatewayConnections[$address])) {
return false;
}
return static::$businessWorker->gatewayConnections[$address]->send($gateway_buffer, true);
}
// 非workerman环境
$gateway_buffer = static::$secretKey ? static::generateAuthBuffer() . $gateway_buffer : $gateway_buffer;
$flag = static::$persistentConnection ? STREAM_CLIENT_PERSISTENT | STREAM_CLIENT_CONNECT : STREAM_CLIENT_CONNECT;
$client = stream_socket_client("tcp://$address", $errno, $errmsg, static::$connectTimeout, $flag);
return strlen($gateway_buffer) == stream_socket_sendto($client, $gateway_buffer);
}
/**
* 向所有 gateway 发送数据
*
* @param string $gateway_data
* @throws Exception
*
* @return void
*/
protected static function sendToAllGateway($gateway_data)
{
$buffer = GatewayProtocol::encode($gateway_data);
// 如果有businessWorker实例,说明运行在workerman环境中,通过businessWorker中的长连接发送数据
if (static::$businessWorker) {
foreach (static::$businessWorker->gatewayConnections as $gateway_connection) {
/** @var TcpConnection $gateway_connection */
$gateway_connection->send($buffer, true);
}
} // 运行在其它环境中,通过注册中心得到gateway地址
else {
$all_addresses = static::getAllGatewayAddressesFromRegister();
foreach ($all_addresses as $address) {
static::sendBufferToGateway($address, $buffer);
}
}
}
/**
* 踢掉某个网关的 socket
*
* @param string $address
* @param int $connection_id
* @return bool
*/
protected static function kickAddress($address, $connection_id, $message)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_KICK;
$gateway_data['connection_id'] = $connection_id;
$gateway_data['body'] = $message;
return static::sendToGateway($address, $gateway_data);
}
/**
* 销毁某个网关的 socket
*
* @param string $address
* @param int $connection_id
* @return bool
*/
protected static function destroyAddress($address, $connection_id)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_DESTROY;
$gateway_data['connection_id'] = $connection_id;
return static::sendToGateway($address, $gateway_data);
}
/**
* 将clientid数组转换成address数组
*
* @param array $client_id_array
* @return array
*/
protected static function clientIdArrayToAddressArray(array $client_id_array)
{
$address_connection_array = array();
foreach ($client_id_array as $client_id) {
$address_data = Context::clientIdToAddress($client_id);
if ($address_data) {
$address = long2ip($address_data['local_ip']) .
":{$address_data['local_port']}";
$address_connection_array[$address][$address_data['connection_id']] = $address_data['connection_id'];
}
}
return $address_connection_array;
}
/**
* 设置 gateway 实例
*
* @param \GatewayWorker\BusinessWorker $business_worker_instance
*/
public static function setBusinessWorker($business_worker_instance)
{
static::$businessWorker = $business_worker_instance;
}
/**
* 获取通过注册中心获取所有 gateway 通讯地址
*
* @return array
* @throws Exception
*/
protected static function getAllGatewayAddressesFromRegister()
{
static $addresses_cache, $last_update;
$time_now = time();
$expiration_time = 1;
$register_addresses = (array)static::$registerAddress;
if(empty($addresses_cache) || $time_now - $last_update > $expiration_time) {
foreach ($register_addresses as $register_address) {
$client = stream_socket_client('tcp://' . $register_address, $errno, $errmsg, static::$connectTimeout);
if (!$client) {
continue;
}
}
if (!$client) {
throw new Exception('Can not connect to tcp://' . $register_address . ' ' . $errmsg);
}
fwrite($client, '{"event":"worker_connect","secret_key":"' . static::$secretKey . '"}' . "\n");
stream_set_timeout($client, 5);
$ret = fgets($client, 655350);
if (!$ret || !$data = json_decode(trim($ret), true)) {
throw new Exception('getAllGatewayAddressesFromRegister fail. tcp://' .
$register_address . ' return ' . var_export($ret, true));
}
$last_update = $time_now;
$addresses_cache = $data['addresses'];
}
if (!$addresses_cache) {
throw new Exception('Gateway::getAllGatewayAddressesFromRegister() with registerAddress:' .
json_encode(static::$registerAddress) . ' return ' . var_export($addresses_cache, true));
}
return $addresses_cache;
}
/**
* 检查群组id是否合法
*
* @param $group
* @return bool
*/
protected static function isValidGroupId($group)
{
if (empty($group)) {
echo new \Exception('group('.var_export($group, true).') empty');
return false;
}
return true;
}
}
if (!class_exists('\Protocols\GatewayProtocol')) {
class_alias('GatewayWorker\Protocols\GatewayProtocol', 'Protocols\GatewayProtocol');
}
@@ -0,0 +1,216 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<[email protected]>
* @copyright walkor<[email protected]>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace GatewayWorker\Protocols;
/**
* Gateway 与 Worker 间通讯的二进制协议
*
* struct GatewayProtocol
* {
* unsigned int pack_len,
* unsigned char cmd,//命令字
* unsigned int local_ip,
* unsigned short local_port,
* unsigned int client_ip,
* unsigned short client_port,
* unsigned int connection_id,
* unsigned char flag,
* unsigned short gateway_port,
* unsigned int ext_len,
* char[ext_len] ext_data,
* char[pack_length-HEAD_LEN] body//包体
* }
* NCNnNnNCnN
*/
class GatewayProtocol
{
// 发给workergateway有一个新的连接
const CMD_ON_CONNECT = 1;
// 发给worker的,客户端有消息
const CMD_ON_MESSAGE = 3;
// 发给worker上的关闭链接事件
const CMD_ON_CLOSE = 4;
// 发给gateway的向单个用户发送数据
const CMD_SEND_TO_ONE = 5;
// 发给gateway的向所有用户发送数据
const CMD_SEND_TO_ALL = 6;
// 发给gateway的踢出用户
// 1、如果有待发消息,将在发送完后立即销毁用户连接
// 2、如果无待发消息,将立即销毁用户连接
const CMD_KICK = 7;
// 发给gateway的立即销毁用户连接
const CMD_DESTROY = 8;
// 发给gateway,通知用户session更新
const CMD_UPDATE_SESSION = 9;
// 获取在线状态
const CMD_GET_ALL_CLIENT_SESSIONS = 10;
// 判断是否在线
const CMD_IS_ONLINE = 11;
// client_id绑定到uid
const CMD_BIND_UID = 12;
// 解绑
const CMD_UNBIND_UID = 13;
// 向uid发送数据
const CMD_SEND_TO_UID = 14;
// 根据uid获取绑定的clientid
const CMD_GET_CLIENT_ID_BY_UID = 15;
// 加入组
const CMD_JOIN_GROUP = 20;
// 离开组
const CMD_LEAVE_GROUP = 21;
// 向组成员发消息
const CMD_SEND_TO_GROUP = 22;
// 获取组成员
const CMD_GET_CLIENT_SESSIONS_BY_GROUP = 23;
// 获取组在线连接数
const CMD_GET_CLIENT_COUNT_BY_GROUP = 24;
// 按照条件查找
const CMD_SELECT = 25;
// 获取在线的群组ID
const CMD_GET_GROUP_ID_LIST = 26;
// 取消分组
const CMD_UNGROUP = 27;
// worker连接gateway事件
const CMD_WORKER_CONNECT = 200;
// 心跳
const CMD_PING = 201;
// GatewayClient连接gateway事件
const CMD_GATEWAY_CLIENT_CONNECT = 202;
// 根据client_id获取session
const CMD_GET_SESSION_BY_CLIENT_ID = 203;
// 发给gateway,覆盖session
const CMD_SET_SESSION = 204;
// 当websocket握手时触发,只有websocket协议支持此命令字
const CMD_ON_WEBSOCKET_CONNECT = 205;
// 包体是标量
const FLAG_BODY_IS_SCALAR = 0x01;
// 通知gateway在send时不调用协议encode方法,在广播组播时提升性能
const FLAG_NOT_CALL_ENCODE = 0x02;
/**
* 包头长度
*
* @var int
*/
const HEAD_LEN = 28;
public static $empty = array(
'cmd' => 0,
'local_ip' => 0,
'local_port' => 0,
'client_ip' => 0,
'client_port' => 0,
'connection_id' => 0,
'flag' => 0,
'gateway_port' => 0,
'ext_data' => '',
'body' => '',
);
/**
* 返回包长度
*
* @param string $buffer
* @return int return current package length
*/
public static function input($buffer)
{
if (strlen($buffer) < self::HEAD_LEN) {
return 0;
}
$data = unpack("Npack_len", $buffer);
return $data['pack_len'];
}
/**
* 获取整个包的 buffer
*
* @param mixed $data
* @return string
*/
public static function encode($data)
{
$flag = (int)is_scalar($data['body']);
if (!$flag) {
$data['body'] = serialize($data['body']);
}
$data['flag'] |= $flag;
$ext_len = strlen($data['ext_data']);
$package_len = self::HEAD_LEN + $ext_len + strlen($data['body']);
return pack("NCNnNnNCnN", $package_len,
$data['cmd'], $data['local_ip'],
$data['local_port'], $data['client_ip'],
$data['client_port'], $data['connection_id'],
$data['flag'], $data['gateway_port'],
$ext_len) . $data['ext_data'] . $data['body'];
}
/**
* 从二进制数据转换为数组
*
* @param string $buffer
* @return array
*/
public static function decode($buffer)
{
$data = unpack("Npack_len/Ccmd/Nlocal_ip/nlocal_port/Nclient_ip/nclient_port/Nconnection_id/Cflag/ngateway_port/Next_len",
$buffer);
if ($data['ext_len'] > 0) {
$data['ext_data'] = substr($buffer, self::HEAD_LEN, $data['ext_len']);
if ($data['flag'] & self::FLAG_BODY_IS_SCALAR) {
$data['body'] = substr($buffer, self::HEAD_LEN + $data['ext_len']);
} else {
$data['body'] = unserialize(substr($buffer, self::HEAD_LEN + $data['ext_len']));
}
} else {
$data['ext_data'] = '';
if ($data['flag'] & self::FLAG_BODY_IS_SCALAR) {
$data['body'] = substr($buffer, self::HEAD_LEN);
} else {
$data['body'] = unserialize(substr($buffer, self::HEAD_LEN));
}
}
return $data;
}
}
@@ -0,0 +1,190 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<[email protected]>
* @copyright walkor<[email protected]>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace GatewayWorker;
use Workerman\Worker;
use Workerman\Lib\Timer;
/**
*
* 注册中心,用于注册 Gateway 和 BusinessWorker
*
* @author walkor<[email protected]>
*
*/
class Register extends Worker
{
/**
* {@inheritdoc}
*/
public $name = 'Register';
/**
* {@inheritdoc}
*/
public $reloadable = false;
/**
* 秘钥
* @var string
*/
public $secretKey = '';
/**
* 所有 gateway 的连接
*
* @var array
*/
protected $_gatewayConnections = array();
/**
* 所有 worker 的连接
*
* @var array
*/
protected $_workerConnections = array();
/**
* 进程启动时间
*
* @var int
*/
protected $_startTime = 0;
/**
* {@inheritdoc}
*/
public function run()
{
// 设置 onMessage 连接回调
$this->onConnect = array($this, 'onConnect');
// 设置 onMessage 回调
$this->onMessage = array($this, 'onMessage');
// 设置 onClose 回调
$this->onClose = array($this, 'onClose');
// 记录进程启动的时间
$this->_startTime = time();
// 强制使用text协议
$this->protocol = '\Workerman\Protocols\Text';
// 运行父方法
parent::run();
}
/**
* 设置个定时器,将未及时发送验证的连接关闭
*
* @param \Workerman\Connection\ConnectionInterface $connection
* @return void
*/
public function onConnect($connection)
{
$connection->timeout_timerid = Timer::add(10, function () use ($connection) {
Worker::log("Register auth timeout (".$connection->getRemoteIp()."). See http://doc2.workerman.net/register-auth-timeout.html");
$connection->close();
}, null, false);
}
/**
* 设置消息回调
*
* @param \Workerman\Connection\ConnectionInterface $connection
* @param string $buffer
* @return void
*/
public function onMessage($connection, $buffer)
{
// 删除定时器
Timer::del($connection->timeout_timerid);
$data = @json_decode($buffer, true);
if (empty($data['event'])) {
$error = "Bad request for Register service. Request info(IP:".$connection->getRemoteIp().", Request Buffer:$buffer). See http://doc2.workerman.net/register-auth-timeout.html";
Worker::log($error);
return $connection->close($error);
}
$event = $data['event'];
$secret_key = isset($data['secret_key']) ? $data['secret_key'] : '';
// 开始验证
switch ($event) {
// 是 gateway 连接
case 'gateway_connect':
if (empty($data['address'])) {
echo "address not found\n";
return $connection->close();
}
if ($secret_key !== $this->secretKey) {
Worker::log("Register: Key does not match ".var_export($secret_key, true)." !== ".var_export($this->secretKey, true));
return $connection->close();
}
$this->_gatewayConnections[$connection->id] = $data['address'];
$this->broadcastAddresses();
break;
// 是 worker 连接
case 'worker_connect':
if ($secret_key !== $this->secretKey) {
Worker::log("Register: Key does not match ".var_export($secret_key, true)." !== ".var_export($this->secretKey, true));
return $connection->close();
}
$this->_workerConnections[$connection->id] = $connection;
$this->broadcastAddresses($connection);
break;
case 'ping':
break;
default:
Worker::log("Register unknown event:$event IP: ".$connection->getRemoteIp()." Buffer:$buffer. See http://doc2.workerman.net/register-auth-timeout.html");
$connection->close();
}
}
/**
* 连接关闭时
*
* @param \Workerman\Connection\ConnectionInterface $connection
*/
public function onClose($connection)
{
if (isset($this->_gatewayConnections[$connection->id])) {
unset($this->_gatewayConnections[$connection->id]);
$this->broadcastAddresses();
}
if (isset($this->_workerConnections[$connection->id])) {
unset($this->_workerConnections[$connection->id]);
}
}
/**
* 向 BusinessWorker 广播 gateway 内部通讯地址
*
* @param \Workerman\Connection\ConnectionInterface $connection
*/
public function broadcastAddresses($connection = null)
{
$data = array(
'event' => 'broadcast_addresses',
'addresses' => array_unique(array_values($this->_gatewayConnections)),
);
$buffer = json_encode($data);
if ($connection) {
$connection->send($buffer);
return;
}
foreach ($this->_workerConnections as $con) {
$con->send($buffer);
}
}
}