diff --git a/composer.json b/composer.json index 85aeec66..7b0b8bce 100644 --- a/composer.json +++ b/composer.json @@ -21,7 +21,6 @@ ], "require": { "php": ">=7.2", - "swoole/ide-helper": "@dev", "ext-mbstring": "*", "doctrine/annotations": "~1.10", "ext-json": "*", @@ -60,12 +59,17 @@ ] }, "require-dev": { - "phpunit/phpunit": "^9.3" + "phpunit/phpunit": "^9.3", + "swoole/ide-helper": "@dev" }, "repositories": [ { "type": "path", "url": "/Users/jerry/project/git-project/zhamao-console" + }, + { + "type": "path", + "url": "/Users/jerry/project/git-project/zhamao-lock" } ] } diff --git a/config/console_color.json b/config/console_color.json index e398759f..ffb81891 100644 --- a/config/console_color.json +++ b/config/console_color.json @@ -10,11 +10,20 @@ }, "white-term": { "success": "green", - "info": "black", + "info": "", "warning": "yellow", "error": "red", "verbose": "blue", "debug": "gray", "trace": "gray" + }, + "no-color": { + "success": "", + "info": "", + "warning": "", + "error": "", + "verbose": "", + "debug": "", + "trace": "" } } diff --git a/config/global.php b/config/global.php index c594efbc..83379206 100644 --- a/config/global.php +++ b/config/global.php @@ -27,20 +27,20 @@ $config['crash_dir'] = $config['zm_data'] . 'crash/'; /** 对应swoole的server->set参数 */ $config['swoole'] = [ 'log_file' => $config['crash_dir'] . 'swoole_error.log', - 'worker_num' => 8, + 'worker_num' => swoole_cpu_num(), 'dispatch_mode' => 2, - 'max_coroutine' => 30000, + 'max_coroutine' => 300000, //'task_worker_num' => 4, //'task_enable_coroutine' => true ]; /** 轻量字符串缓存，默认开启 */ $config['light_cache'] = [ - "status" => true, - "size" => 2048, //最多允许储存的条数（需要2的倍数） - "max_strlen" => 4096, //单行字符串最大长度（需要2的倍数） + "size" => 1024, //最多允许储存的条数（需要2的倍数） + "max_strlen" => 8192, //单行字符串最大长度（需要2的倍数） "hash_conflict_proportion" => 0.6, //Hash冲突率（越大越好，但是需要的内存更多） - "persistence_path" => $config['zm_data']."_cache.json" + "persistence_path" => $config['zm_data']."_cache.json", + 'auto_save_interval' => 900 ]; /** MySQL数据库连接信息，host留空则启动时不创建sql连接池 */ @@ -60,8 +60,17 @@ $config['sql_config'] = [ 'sql_default_fetch_mode' => PDO::FETCH_ASSOC // added in 1.5.6 ]; -/** CQHTTP连接约定的token */ -$config["access_token"] = ""; +/** Redis连接信息，host留空则启动时不创建Redis连接池 */ +$config['redis_config'] = [ + 'host' => '', + 'port' => 6379, + 'timeout' => 1, + 'db_index' => 0, + 'auth' => '' +]; + +/** onebot连接约定的token */ +$config["access_token"] = ''; /** HTTP服务器固定请求头的返回 */ $config['http_header'] = [ @@ -82,9 +91,6 @@ $config['init_atomics'] = [ /** 终端日志显示等级（0-4） */ $config["info_level"] = 2; -/** 自动保存计时器的缓存保存时间（秒） */ -$config['auto_save_interval'] = 900; - /** 上下文接口类 implemented from ContextInterface */ $config['context_class'] = \ZM\Context\Context::class; @@ -108,7 +114,7 @@ $config['command_register_class'] = [ ]; /** 服务器启用的外部第三方和内部插件 */ -$config['plugins'] = [ +$config['modules'] = [ 'qqbot' => true, // QQ机器人事件解析器，如果取消此项则默认为 true 开启状态，否则你手动填写 false 才会关闭 ]; diff --git a/src/Module/Example/Hello.php b/src/Module/Example/Hello.php index db3b718c..ad387249 100644 --- a/src/Module/Example/Hello.php +++ b/src/Module/Example/Hello.php @@ -2,16 +2,15 @@ namespace Module\Example; +use ZM\Annotation\CQ\CQMetaEvent; +use ZM\Annotation\Swoole\OnSwooleEvent; use ZM\Annotation\Swoole\OnWorkerStart; -use ZM\Annotation\Swoole\OnTick; use ZM\ConnectionManager\ConnectionObject; use ZM\Console\Console; use ZM\Annotation\CQ\CQCommand; -use ZM\Annotation\Http\Middleware; use ZM\Annotation\Http\RequestMapping; -use ZM\Annotation\Swoole\SwooleEvent; use ZM\Store\LightCache; -use ZM\Store\ZMBuf; +use ZM\Store\Lock\SpinLock; use ZM\Utils\ZMUtil; /** @@ -23,7 +22,7 @@ class Hello { /** * 在机器人连接后向终端输出信息 - * @SwooleEvent("open",rule="connectIsQQ()") + * @OnSwooleEvent("open",rule="connectIsQQ()") * @param $conn */ public function onConnect(ConnectionObject $conn) { @@ -32,7 +31,7 @@ class Hello /** * 在机器人断开连接后向终端输出信息 - * @SwooleEvent("close",rule="connectIsQQ()") + * @OnSwooleEvent("close",rule="connectIsQQ()") * @param ConnectionObject $conn */ public function onDisconnect(ConnectionObject $conn) { @@ -59,6 +58,7 @@ class Hello * @CQCommand("随机数") * @CQCommand(regexMatch="*从*到*的随机数") * @param $arg + * @return string */ public function randNum($arg) { // 获取第一个数字类型的参数 @@ -68,7 +68,7 @@ class Hello $a = min(intval($num1), intval($num2)); $b = max(intval($num1), intval($num2)); // 回复用户结果 - ctx()->reply("随机数是：" . mt_rand($a, $b)); + return "随机数是：" . mt_rand($a, $b); } /** @@ -76,7 +76,6 @@ class Hello * @RequestMapping("/httpTimer") */ public function timer() { - ZMBuf::atomic("_tmp_2")->add(1); return "This page is used as testing TimerMiddleware! Do not use it in production."; } @@ -101,7 +100,7 @@ class Hello /** * 框架会默认关闭未知的WebSocket链接，因为这个绑定的事件，你可以根据你自己的需求进行修改 - * @SwooleEvent(type="open",rule="connectIsDefault()") + * @OnSwooleEvent(type="open",rule="connectIsDefault()") */ public function closeUnknownConn() { Console::info("Unknown connection , I will close it."); diff --git a/src/ZM/API/CQ.php b/src/ZM/API/CQ.php index 346fa8ce..61191446 100644 --- a/src/ZM/API/CQ.php +++ b/src/ZM/API/CQ.php @@ -5,7 +5,6 @@ namespace ZM\API; use ZM\Console\Console; -use ZM\Utils\ZMUtil; class CQ { diff --git a/src/ZM/API/CQAPI.php b/src/ZM/API/CQAPI.php index 751a364a..d45540fd 100644 --- a/src/ZM/API/CQAPI.php +++ b/src/ZM/API/CQAPI.php @@ -6,9 +6,9 @@ namespace ZM\API; use Co; use ZM\ConnectionManager\ConnectionObject; use ZM\Console\Console; -use ZM\Event\EventHandler; -use ZM\Store\LightCache; -use ZM\Store\ZMBuf; +use ZM\Store\LightCacheInside; +use ZM\Store\Lock\SpinLock; +use ZM\Store\ZMAtomic; trait CQAPI { @@ -28,30 +28,28 @@ trait CQAPI } public function processWebsocketAPI($connection, $reply, $function = false) { - $api_id = ZMBuf::atomic("wait_msg_id")->get(); + $api_id = ZMAtomic::get("wait_msg_id")->add(1); $reply["echo"] = $api_id; - ZMBuf::atomic("wait_msg_id")->add(1); - EventHandler::callCQAPISend($reply, $connection); - if ($function === true) { - LightCache::set("sent_api_".$api_id, [ - "data" => $reply, - "time" => microtime(true), - "coroutine" => Co::getuid(), - "self_id" => $connection->getOption("connect_id") - ]); - } else { - LightCache::set("sent_api_".$api_id, [ - "data" => $reply, - "time" => microtime(true), - "self_id" => $connection->getOption("connect_id") - ]); - } - + //EventHandler::callCQAPISend($reply, $connection); + SpinLock::lock("wait_api"); + $r = LightCacheInside::get("wait_api", "wait_api"); + $r[$api_id] = [ + "data" => $reply, + "time" => microtime(true), + "self_id" => $connection->getOption("connect_id") + ]; + if ($function === true) $r[$api_id]["coroutine"] = Co::getuid(); + LightCacheInside::set("wait_api", "wait_api", $r); + SpinLock::unlock("wait_api"); if (server()->push($connection->getFd(), json_encode($reply))) { if ($function === true) { Co::suspend(); - $data = LightCache::get("sent_api_".$api_id); - LightCache::unset("sent_api_".$api_id); + SpinLock::lock("wait_api"); + $r = LightCacheInside::get("wait_api", "wait_api"); + $data = $r[$api_id]; + unset($r[$api_id]); + LightCacheInside::set("wait_api", "wait_api", $r); + SpinLock::unlock("wait_api"); return isset($data['result']) ? $data['result'] : null; } return true; @@ -63,15 +61,23 @@ trait CQAPI "data" => null, "self_id" => $connection->getOption("connect_id") ]; - $s = LightCache::get("sent_api_".$reply["echo"]); - if (($s["func"] ?? null) !== null) - call_user_func($s["func"], $response, $reply); - LightCache::unset("sent_api_".$reply["echo"]); + SpinLock::lock("wait_api"); + $r = LightCacheInside::get("wait_api", "wait_api"); + unset($r[$reply["echo"]]); + LightCacheInside::set("wait_api", "wait_api", $r); + SpinLock::unlock("wait_api"); if ($function === true) return $response; return false; } } + /** + * @param $connection + * @param $reply + * @param null $function + * @return bool + * @noinspection PhpUnusedParameterInspection + */ public function processHttpAPI($connection, $reply, $function = null) { return false; } diff --git a/src/ZM/Annotation/AnnotationParser.php b/src/ZM/Annotation/AnnotationParser.php index e5f40070..6bcb7eaf 100644 --- a/src/ZM/Annotation/AnnotationParser.php +++ b/src/ZM/Annotation/AnnotationParser.php @@ -201,6 +201,7 @@ class AnnotationParser $array[0]['method'] = $method; $array[0]['class'] = $class; $array[0]['request_method'] = $vss->request_method; + $array[0]['route'] = $vss->route; $this->req_mapping = $array; return; } diff --git a/src/ZM/Annotation/Swoole/SwooleEvent.php b/src/ZM/Annotation/Swoole/OnSwooleEvent.php similarity index 93% rename from src/ZM/Annotation/Swoole/SwooleEvent.php rename to src/ZM/Annotation/Swoole/OnSwooleEvent.php index 105125a4..d71f467b 100644 --- a/src/ZM/Annotation/Swoole/SwooleEvent.php +++ b/src/ZM/Annotation/Swoole/OnSwooleEvent.php @@ -10,12 +10,12 @@ use ZM\Annotation\Interfaces\Level; use ZM\Annotation\Interfaces\Rule; /** - * Class SwooleEvent + * Class OnSwooleEvent * @Annotation * @Target("ALL") * @package ZM\Annotation\Swoole */ -class SwooleEvent extends AnnotationBase implements Rule, Level +class OnSwooleEvent extends AnnotationBase implements Rule, Level { /** * @var string diff --git a/src/ZM/Command/PureHttpCommand.php b/src/ZM/Command/PureHttpCommand.php index d30ce4be..99c9114c 100644 --- a/src/ZM/Command/PureHttpCommand.php +++ b/src/ZM/Command/PureHttpCommand.php @@ -17,7 +17,7 @@ use Symfony\Component\Console\Input\InputOption; use Symfony\Component\Console\Output\OutputInterface; use ZM\Config\ZMConfig; use ZM\Console\Console; -use ZM\Store\ZMBuf; +use ZM\Store\ZMAtomic; use ZM\Utils\HttpUtil; class PureHttpCommand extends Command @@ -28,7 +28,7 @@ class PureHttpCommand extends Command protected function configure() { $this->setDescription("Run a simple http server | 启动一个简单的文件 HTTP 服务器"); $this->setHelp("直接运行可以启动"); - $this->addArgument('dir', InputArgument::OPTIONAL, 'Your directory'); + $this->addArgument('dir', InputArgument::REQUIRED, 'Your directory'); $this->addOption("host", 'H', InputOption::VALUE_REQUIRED, "启动监听地址"); $this->addOption("port", 'P', InputOption::VALUE_REQUIRED, "启动监听地址的端口"); // ... @@ -36,19 +36,23 @@ class PureHttpCommand extends Command protected function execute(InputInterface $input, OutputInterface $output) { $tty_width = explode(" ", trim(exec("stty size")))[1]; + if(realpath($input->getArgument('dir') ?? '.') === false) { + $output->writeln("Directory error(".($input->getArgument('dir') ?? '.')."): no such file or directory."); + return self::FAILURE; + } $global = ZMConfig::get("global"); $host = $input->getOption("host") ?? $global["host"]; $port = $input->getOption("port") ?? $global["port"]; $server = new Server($host, $port); $server->set(ZMConfig::get("global", "swoole")); Console::init(0, $server); - ZMBuf::$atomics["request"] = []; + ZMAtomic::$atomics["request"] = []; for ($i = 0; $i < 32; ++$i) { - ZMBuf::$atomics["request"][$i] = new Atomic(0); + ZMAtomic::$atomics["request"][$i] = new Atomic(0); } $index = ["index.html", "index.htm"]; $server->on("request", function (Request $request, Response $response) use ($input, $index, $server) { - ZMBuf::$atomics["request"][$server->worker_id]->add(1); + ZMAtomic::$atomics["request"][$server->worker_id]->add(1); HttpUtil::handleStaticPage( $request->server["request_uri"], $response, @@ -62,7 +66,7 @@ class PureHttpCommand extends Command Process::signal(SIGINT, function () use ($server) { Console::warning("Server interrupted by keyboard."); for ($i = 0; $i < 32; ++$i) { - $num = ZMBuf::$atomics["request"][$i]->get(); + $num = ZMAtomic::$atomics["request"][$i]->get(); if($num != 0) echo "[$i]: ".$num."\n"; } diff --git a/src/ZM/Command/RunServerCommand.php b/src/ZM/Command/RunServerCommand.php index 9ffd6614..2f063f8a 100644 --- a/src/ZM/Command/RunServerCommand.php +++ b/src/ZM/Command/RunServerCommand.php @@ -15,19 +15,23 @@ class RunServerCommand extends Command protected static $defaultName = 'server'; protected function configure() { + $this->setDefinition([ + new InputOption("debug-mode", "D", null, "开启调试模式 (这将关闭协程化)"), + new InputOption("log-debug", null, null, "调整消息等级到debug (log-level=4)"), + new InputOption("log-verbose", null, null, "调整消息等级到verbose (log-level=3)"), + new InputOption("log-info", null, null, "调整消息等级到info (log-level=2)"), + new InputOption("log-warning", null, null, "调整消息等级到warning (log-level=1)"), + new InputOption("log-error", null, null, "调整消息等级到error (log-level=0)"), + new InputOption("log-theme", null, InputOption::VALUE_REQUIRED, "改变终端的主题配色"), + new InputOption("disable-console-input", null, null, "禁止终端输入内容 (后台服务时需要)"), + new InputOption("disable-coroutine", null, null, "关闭协程Hook"), + new InputOption("daemon", null, null, "以守护进程的方式运行框架"), + new InputOption("watch", null, null, "监听 src/ 目录的文件变化并热更新"), + new InputOption("env", null, InputOption::VALUE_REQUIRED, "设置环境类型 (production, development, staging)"), + ]); $this->setDescription("Run zhamao-framework | 启动框架"); $this->setHelp("直接运行可以启动"); - $this->addOption("debug-mode", "D", null, "开启调试模式 (这将关闭协程化)"); - $this->addOption("log-debug", null, null, "调整消息等级到debug (log-level=4)"); - $this->addOption("log-verbose", null, null, "调整消息等级到verbose (log-level=3)"); - $this->addOption("log-info", null, null, "调整消息等级到info (log-level=2)"); - $this->addOption("log-warning", null, null, "调整消息等级到warning (log-level=1)"); - $this->addOption("log-error", null, null, "调整消息等级到error (log-level=0)"); - $this->addOption("log-theme", null, InputOption::VALUE_REQUIRED, "改变终端的主题配色"); - $this->addOption("disable-console-input", null, null, "禁止终端输入内容 (后台服务时需要)"); - $this->addOption("disable-coroutine", null, null, "关闭协程Hook"); - $this->addOption("watch", null, null, "监听 src/ 目录的文件变化并热更新"); - $this->addOption("env", null, InputOption::VALUE_REQUIRED, "设置环境类型 (production, development, staging)"); + // ... } diff --git a/src/ZM/ConsoleApplication.php b/src/ZM/ConsoleApplication.php index 67ea073d..dd6bb32b 100644 --- a/src/ZM/ConsoleApplication.php +++ b/src/ZM/ConsoleApplication.php @@ -31,7 +31,6 @@ class ConsoleApplication extends Application new PureHttpCommand() ]); //if (LOAD_MODE === 0) $this->add(new BuildCommand()); //只有在git源码模式才能使用打包指令 - if (LOAD_MODE === 0) define("WORKING_DIR", getcwd()); elseif (LOAD_MODE == 1) define("WORKING_DIR", realpath(__DIR__ . "/../../")); elseif (LOAD_MODE == 2) echo "Phar mode: " . WORKING_DIR . PHP_EOL; diff --git a/src/ZM/Context/Context.php b/src/ZM/Context/Context.php index f7d0ed0e..f721fdd2 100644 --- a/src/ZM/Context/Context.php +++ b/src/ZM/Context/Context.php @@ -14,8 +14,10 @@ use ZM\Exception\InvalidArgumentException; use ZM\Exception\WaitTimeoutException; use ZM\Http\Response; use ZM\API\ZMRobot; -use ZM\Store\LightCache; -use ZM\Store\ZMBuf; +use ZM\Store\LightCacheInside; +use ZM\Store\Lock\SpinLock; +use ZM\Store\ZMAtomic; +use ZM\Utils\CoMessage; class Context implements ContextInterface { @@ -137,12 +139,19 @@ class Context implements ContextInterface * @throws WaitTimeoutException */ public function waitMessage($prompt = "", $timeout = 600, $timeout_prompt = "") { - if ($prompt != "") $this->reply($prompt); if (!isset($this->getData()["user_id"], $this->getData()["message"], $this->getData()["self_id"])) throw new InvalidArgumentException("协程等待参数缺失"); + + + if ($prompt != "") $this->reply($prompt); + + $r = CoMessage::yieldByWS($this->getData(), ["user_id", "self_id", "message_type", onebot_target_id_name($this->getMessageType())]); + if($r === false) { + throw new WaitTimeoutException($this, $timeout_prompt); + } + $cid = Co::getuid(); - $api_id = ZMBuf::atomic("wait_msg_id")->get(); - ZMBuf::atomic("wait_msg_id")->add(1); + $api_id = ZMAtomic::get("wait_msg_id")->add(1); $hang = [ "coroutine" => $cid, "user_id" => $this->getData()["user_id"], @@ -154,17 +163,24 @@ class Context implements ContextInterface if ($hang["message_type"] == "group" || $hang["message_type"] == "discuss") { $hang[$hang["message_type"] . "_id"] = $this->getData()[$this->getData()["message_type"] . "_id"]; } - LightCache::set("wait_api_".$api_id, $hang); + SpinLock::lock("wait_api"); + $hw = LightCacheInside::get("wait_api", "wait_api") ?? []; + $hw[$api_id] = $hang; + LightCacheInside::set("wait_api", "wait_api", $hw); + SpinLock::unlock("wait_api"); $id = swoole_timer_after($timeout * 1000, function () use ($api_id, $timeout_prompt) { - $r = LightCache::get("wait_api_".$api_id); + $r = LightCacheInside::get("wait_api", "wait_api")[$api_id] ?? null; if (is_array($r)) { Co::resume($r["coroutine"]); } }); Co::suspend(); - $sess = LightCache::get("wait_api_".$api_id); - LightCache::unset("wait_api_".$api_id); + SpinLock::lock("wait_api"); + $hw = LightCacheInside::get("wait_api", "wait_api") ?? []; + $sess = $hw[$api_id]; + unset($hw[$api_id]); + LightCacheInside::set("wait_api", "wait_api", $hw); $result = $sess["result"]; if (isset($id)) swoole_timer_clear($id); if ($result === null) throw new WaitTimeoutException($this, $timeout_prompt); diff --git a/src/ZM/DB/DB.php b/src/ZM/DB/DB.php index 9163c44d..0dd1c35d 100644 --- a/src/ZM/DB/DB.php +++ b/src/ZM/DB/DB.php @@ -7,7 +7,7 @@ namespace ZM\DB; use Exception; use ZM\Config\ZMConfig; use ZM\Console\Console; -use ZM\Store\ZMBuf; +use ZM\Store\MySQL\SqlPoolStorage; use PDOException; use PDOStatement; use Swoole\Database\PDOStatementProxy; @@ -35,11 +35,11 @@ class DB * @return Table * @throws DbException */ - public static function table($table_name, $enable_cache = null) { + public static function table($table_name) { if (Table::getTableInstance($table_name) === null) { if (in_array($table_name, self::$table_list)) - return new Table($table_name, $enable_cache ?? ZMConfig::get("global", "sql_config")["sql_enable_cache"]); - elseif(ZMBuf::$sql_pool !== null){ + return new Table($table_name); + elseif(SqlPoolStorage::$sql_pool !== null){ throw new DbException("Table " . $table_name . " not exist in database."); } else { throw new DbException("Database connection not exist or connect failed. Please check sql configuration"); @@ -63,13 +63,13 @@ class DB */ public static function unprepared($line) { try { - $conn = ZMBuf::$sql_pool->get(); + $conn = SqlPoolStorage::$sql_pool->get(); if ($conn === false) { - ZMBuf::$sql_pool->put(null); + SqlPoolStorage::$sql_pool->put(null); throw new DbException("无法连接SQL！" . $line); } $result = $conn->query($line) === false ? false : true; - ZMBuf::$sql_pool->put($conn); + SqlPoolStorage::$sql_pool->put($conn); return $result; } catch (DBException $e) { Console::warning($e->getMessage()); @@ -87,19 +87,19 @@ class DB public static function rawQuery(string $line, $params = [], $fetch_mode = ZM_DEFAULT_FETCH_MODE) { Console::debug("MySQL: ".$line." | ". implode(", ", $params)); try { - $conn = ZMBuf::$sql_pool->get(); + $conn = SqlPoolStorage::$sql_pool->get(); if ($conn === false) { - ZMBuf::$sql_pool->put(null); + SqlPoolStorage::$sql_pool->put(null); throw new DbException("无法连接SQL！" . $line); } $ps = $conn->prepare($line); if ($ps === false) { - ZMBuf::$sql_pool->put(null); + SqlPoolStorage::$sql_pool->put(null); throw new DbException("SQL语句查询错误，" . $line . "，错误信息：" . $conn->error); } else { if (!($ps instanceof PDOStatement) && !($ps instanceof PDOStatementProxy)) { var_dump($ps); - ZMBuf::$sql_pool->put(null); + SqlPoolStorage::$sql_pool->put(null); throw new DbException("语句查询错误！返回的不是 PDOStatement" . $line); } if ($params == []) $result = $ps->execute(); @@ -107,11 +107,11 @@ class DB $result = $ps->execute([$params]); } else $result = $ps->execute($params); if ($result !== true) { - ZMBuf::$sql_pool->put(null); + SqlPoolStorage::$sql_pool->put(null); throw new DBException("语句[$line]错误！" . $ps->errorInfo()[2]); //echo json_encode(debug_backtrace(), 128 | 256); } - ZMBuf::$sql_pool->put($conn); + SqlPoolStorage::$sql_pool->put($conn); return $ps->fetchAll($fetch_mode); } } catch (DbException $e) { diff --git a/src/ZM/DB/SelectBody.php b/src/ZM/DB/SelectBody.php index a7c882af..903916db 100644 --- a/src/ZM/DB/SelectBody.php +++ b/src/ZM/DB/SelectBody.php @@ -4,7 +4,6 @@ namespace ZM\DB; -use ZM\Console\Console; use ZM\Exception\DbException; class SelectBody @@ -46,17 +45,7 @@ class SelectBody * @throws DbException */ public function fetchAll($fetch_mode = ZM_DEFAULT_FETCH_MODE) { - if ($this->table->isCacheEnabled()) { - $rr = md5(implode(",", $this->select_thing) . serialize($this->where_thing)); - if (array_key_exists($rr, $this->table->cache)) { - Console::debug('SQL query cached: ' . $rr); - return $this->table->cache[$rr]->getResult(); - } - } $this->execute($fetch_mode); - if ($this->table->isCacheEnabled() && !in_array($rr, $this->table->cache)) { - $this->table->cache[$rr] = $this; - } return $this->getResult(); } diff --git a/src/ZM/DB/Table.php b/src/ZM/DB/Table.php index cf85d4cd..939bc8ad 100644 --- a/src/ZM/DB/Table.php +++ b/src/ZM/DB/Table.php @@ -14,10 +14,7 @@ class Table private static $table_instance = []; - private $enable_cache; - - public function __construct($table_name, $enable_cache) { - $this->enable_cache = $enable_cache; + public function __construct($table_name) { $this->table_name = $table_name; self::$table_instance[$table_name] = $this; } @@ -75,5 +72,4 @@ class Table */ public function getTableName() { return $this->table_name; } - public function isCacheEnabled() { return $this->enable_cache; } -} \ No newline at end of file +} diff --git a/src/ZM/Event/CQ/MessageEvent.php b/src/ZM/Event/CQ/MessageEvent.php deleted file mode 100644 index d128af8e..00000000 --- a/src/ZM/Event/CQ/MessageEvent.php +++ /dev/null @@ -1,169 +0,0 @@ -data = $data; - $this->connection = $conn_or_response; - $this->circle = $circle; - } - - /** - * @return bool - * @throws AnnotationException - */ - public function onBefore() { - $dispatcher = new EventDispatcher(CQBefore::class . "::message"); - $dispatcher->setRuleFunction(function ($v) { - if($v->level < 200) EventDispatcher::interrupt(); - return true; - }); - $dispatcher->setReturnFunction(function($result){ - if(!$result) EventDispatcher::interrupt(); - }); - $dispatcher->dispatchEvents(); - - foreach (ZMBuf::get("wait_api", []) as $k => $v) { - if(zm_data_hash(ctx()->getData()) == $v["hash"]) { - $v["result"] = context()->getData()["message"]; - ZMBuf::appendKey("wait_api", $k, $v); - Co::resume($v["coroutine"]); - return false; - } - if (context()->getData()["user_id"] == $v["user_id"] && - context()->getData()["self_id"] == $v["self_id"] && - context()->getData()["message_type"] == $v["message_type"] && - (context()->getData()[context()->getData()["message_type"] . "_id"] ?? context()->getData()["user_id"]) == - ($v[$v["message_type"] . "_id"] ?? $v["user_id"])) { - $v["result"] = context()->getData()["message"]; - ZMBuf::appendKey("wait_api", $k, $v); - Co::resume($v["coroutine"]); - return false; - } - } - foreach (EventManager::$events[CQBefore::class]["message"] ?? [] as $v) { - if ($v->level >= 200) continue; - $c = $v->class; - if (ctx()->getCache("level") != 0) continue; - EventHandler::callWithMiddleware( - $c, - $v->method, - ["data" => context()->getData(), "connection" => $this->connection], - [], - function ($r) { - if (!$r) context()->setCache("block_continue", true); - } - ); - if (context()->getCache("block_continue") === true) return false; - } - return true; - } - - /** - * @noinspection PhpRedundantCatchClauseInspection - */ - public function onActivate() { - try { - $word = split_explode(" ", str_replace("\r", "", context()->getMessage())); - if (count(explode("\n", $word[0])) >= 2) { - $enter = explode("\n", context()->getMessage()); - $first = split_explode(" ", array_shift($enter)); - $word = array_merge($first, $enter); - foreach ($word as $k => $v) { - $word[$k] = trim($word[$k]); - } - } - - //分发CQCommand事件 - $dispatcher = new EventDispatcher(CQCommand::class); - $dispatcher->setRuleFunction(function ($v) use ($word) { - if ($v->match == "" && $v->regexMatch == "" && $v->fullMatch == "") return false; - elseif (($v->user_id == 0 || ($v->user_id != 0 && $v->user_id == ctx()->getUserId())) && - ($v->group_id == 0 || ($v->group_id != 0 && $v->group_id == (ctx()->getGroupId() ?? 0))) && - ($v->message_type == '' || ($v->message_type != '' && $v->message_type == ctx()->getMessageType())) - ) { - if (($word[0] != "" && $v->match == $word[0]) || - in_array($word[0], $v->alias) || - ($v->regexMatch != "" && ($args = matchArgs($v->regexMatch, ctx()->getMessage())) !== false) || - ($v->fullMatch != "" && (preg_match("/" . $v->fullMatch . "/u", ctx()->getMessage(), $args)) != 0)) { - return true; - } - } - return false; - }); - $dispatcher->setReturnFunction(function ($result) { - if (is_string($result)) ctx()->reply($result); - EventDispatcher::interrupt(); - }); - $r = $dispatcher->dispatchEvents($word); - if ($r === null) return; - - //分发CQMessage事件 - $msg_dispatcher = new EventDispatcher(CQMessage::class); - $msg_dispatcher->setRuleFunction(function ($v) { - return ($v->message == '' || ($v->message != '' && $v->message == context()->getData()["message"])) && - ($v->user_id == 0 || ($v->user_id != 0 && $v->user_id == context()->getData()["user_id"])) && - ($v->group_id == 0 || ($v->group_id != 0 && $v->group_id == (context()->getData()["group_id"] ?? 0))) && - ($v->message_type == '' || ($v->message_type != '' && $v->message_type == context()->getData()["message_type"])) && - ($v->raw_message == '' || ($v->raw_message != '' && $v->raw_message == context()->getData()["raw_message"])); - }); - $msg_dispatcher->setReturnFunction(function ($result) { - if (is_string($result)) ctx()->reply($result); - }); - $msg_dispatcher->dispatchEvents(ctx()->getMessage()); - } catch (WaitTimeoutException $e) { - $e->module->finalReply($e->getMessage()); - } - } - - /** - * 在调用完事件后执行的 - * @throws AnnotationException - */ - public function onAfter() { - context()->setCache("block_continue", null); - foreach (ZMBuf::$events[CQAfter::class]["message"] ?? [] as $v) { - $c = $v->class; - EventHandler::callWithMiddleware( - $c, - $v->method, - ["data" => context()->getData(), "connection" => $this->connection], - [], - function ($r) { - if (!$r) context()->setCache("block_continue", true); - } - ); - if (context()->getCache("block_continue") === true) return false; - } - return true; - } - - public function hasReply() { - return $this->function_call; - } -} diff --git a/src/ZM/Event/CQ/MetaEvent.php b/src/ZM/Event/CQ/MetaEvent.php deleted file mode 100644 index 78a7671e..00000000 --- a/src/ZM/Event/CQ/MetaEvent.php +++ /dev/null @@ -1,71 +0,0 @@ -data = $data; - $this->connection = $connection; - $this->circle = $circle; - } - - /** - * @return bool - * @throws AnnotationException - */ - public function onBefore() { - foreach (ZMBuf::$events[CQBefore::class]["meta_event"] ?? [] as $v) { - $c = $v->class; - EventHandler::callWithMiddleware( - $c, - $v->method, - ["data" => context()->getData(), "connection" => $this->connection], - [], - function ($r) { - if(!$r) context()->setCache("block_continue", true); - } - ); - if(context()->getCache("block_continue") === true) return false; - } - return true; - } - - /** - * @throws AnnotationException - */ - public function onActivate() { - try { - $obj = []; - foreach (ZMBuf::$events[CQMetaEvent::class] ?? [] as $v) { - /** @var CQMetaEvent $v */ - if ( - ($v->meta_event_type == '' || ($v->meta_event_type != '' && $v->meta_event_type == $this->data["meta_event_type"])) && - ($v->sub_type == 0 || ($v->sub_type != 0 && $v->sub_type == $this->data["sub_type"]))) { - $c = $v->class; - if (!isset($obj[$c])) - $obj[$c] = new $c(); - EventHandler::callWithMiddleware($obj[$c],$v->method, [], [], function($r) { - if (is_string($r)) context()->reply($r); - }); - if (context()->getCache("block_continue") === true) return; - } - } - } /** @noinspection PhpRedundantCatchClauseInspection */ catch (WaitTimeoutException $e) { - $e->module->finalReply($e->getMessage()); - } - } -} diff --git a/src/ZM/Event/CQ/NoticeEvent.php b/src/ZM/Event/CQ/NoticeEvent.php deleted file mode 100644 index 4ff05dcf..00000000 --- a/src/ZM/Event/CQ/NoticeEvent.php +++ /dev/null @@ -1,95 +0,0 @@ -data = $data; - $this->connection = $connection; - $this->circle = $circle; - } - - /** - * @return bool - * @throws AnnotationException - */ - public function onBefore() { - foreach (ZMBuf::$events[CQBefore::class]["notice"] ?? [] as $v) { - $c = $v->class; - EventHandler::callWithMiddleware( - $c, - $v->method, - ["data" => context()->getData(), "connection" => $this->connection], - [], - function ($r) { - if(!$r) context()->setCache("block_continue", true); - } - ); - if(context()->getCache("block_continue") === true) return false; - } - return true; - } - - /** - * @throws AnnotationException - */ - public function onActivate() { - try { - $obj = []; - foreach (ZMBuf::$events[CQNotice::class] ?? [] as $v) { - /** @var CQNotice $v */ - if ( - ($v->notice_type == '' || ($v->notice_type != '' && $v->notice_type == $this->data["notice_type"])) && - ($v->sub_type == 0 || ($v->sub_type != 0 && $v->sub_type == $this->data["sub_type"])) && - ($v->group_id == 0 || ($v->group_id != 0 && $v->group_id == ($this->data["group_id"] ?? 0))) && - ($v->operator_id == 0 || ($v->operator_id != 0 && $v->operator_id == ($this->data["operator_id"] ?? 0)))) { - $c = $v->class; - if (!isset($obj[$c])) - $obj[$c] = new $c(); - EventHandler::callWithMiddleware($obj[$c],$v->method, [], [], function($r) { - if (is_string($r)) context()->reply($r); - }); - if (context()->getCache("block_continue") === true) return; - } - } - } /** @noinspection PhpRedundantCatchClauseInspection */ catch (WaitTimeoutException $e) { - $e->module->finalReply($e->getMessage()); - } - } - - /** - * @return bool - * @throws AnnotationException - */ - public function onAfter() { - foreach (ZMBuf::$events[CQAfter::class]["notice"] ?? [] as $v) { - $c = $v->class; - EventHandler::callWithMiddleware( - $c, - $v->method, - ["data" => context()->getData(), "connection" => $this->connection], - [], - function ($r) { - if(!$r) context()->setCache("block_continue", true); - } - ); - if(context()->getCache("block_continue") === true) return false; - } - return true; - } -} diff --git a/src/ZM/Event/CQ/RequestEvent.php b/src/ZM/Event/CQ/RequestEvent.php deleted file mode 100644 index c600fb77..00000000 --- a/src/ZM/Event/CQ/RequestEvent.php +++ /dev/null @@ -1,96 +0,0 @@ -data = $data; - $this->connection = $connection; - $this->circle = $circle; - } - - /** - * @return bool - * @throws AnnotationException - */ - public function onBefore() { - foreach (ZMBuf::$events[CQBefore::class]["request"] ?? [] as $v) { - $c = $v->class; - EventHandler::callWithMiddleware( - $c, - $v->method, - ["data" => context()->getData(), "connection" => $this->connection], - [], - function ($r) { - if(!$r) context()->setCache("block_continue", true); - } - ); - if(context()->getCache("block_continue") === true) return false; - } - return true; - } - - /** - * @throws AnnotationException - * @noinspection PhpRedundantCatchClauseInspection - */ - public function onActivate() { - try { - $obj = []; - foreach (ZMBuf::$events[CQRequest::class] ?? [] as $v) { - /** @var CQRequest $v */ - if ( - ($v->request_type == '' || ($v->request_type != '' && $v->request_type == $this->data["request_type"])) && - ($v->sub_type == 0 || ($v->sub_type != 0 && $v->sub_type == $this->data["sub_type"])) && - ($v->user_id == 0 || ($v->user_id != 0 && $v->user_id == ($this->data["user_id"] ?? 0))) && - ($v->comment == 0 || ($v->comment != 0 && $v->comment == ($this->data["comment"] ?? 0)))) { - $c = $v->class; - if (!isset($obj[$c])) - $obj[$c] = new $c(); - EventHandler::callWithMiddleware($obj[$c],$v->method, [], [], function($r) { - if (is_string($r)) context()->reply($r); - }); - if (context()->getCache("block_continue") === true) return; - } - } - } catch (WaitTimeoutException $e) { - $e->module->finalReply($e->getMessage()); - } - } - - /** - * @return bool - * @throws AnnotationException - */ - public function onAfter() { - foreach (ZMBuf::$events[CQAfter::class]["request"] ?? [] as $v) { - $c = $v->class; - EventHandler::callWithMiddleware( - $c, - $v->method, - ["data" => context()->getData(), "connection" => $this->connection], - [], - function ($r) { - if(!$r) context()->setCache("block_continue", true); - } - ); - if(context()->getCache("block_continue") === true) return false; - } - return true; - } -} diff --git a/src/ZM/Event/Event.php b/src/ZM/Event/Event.php deleted file mode 100644 index f78aff33..00000000 --- a/src/ZM/Event/Event.php +++ /dev/null @@ -1,11 +0,0 @@ -class] ?? [] as $v) { + + foreach ((EventManager::$events[$this->class] ?? []) as $v) { + if($this->class == CQMetaEvent::class) { + //eval(BP); + } $result = $this->dispatchEvent($v, $this->rule, ...$params); - if (is_callable($this->return_func)) ($this->return_func)($result); + if ($result !== false && is_callable($this->return_func)) ($this->return_func)($result); } return true; } catch (InterruptException $e) { diff --git a/src/ZM/Event/EventHandler.php b/src/ZM/Event/EventHandler.php deleted file mode 100644 index 6636d2e5..00000000 --- a/src/ZM/Event/EventHandler.php +++ /dev/null @@ -1,293 +0,0 @@ -fd); - set_coroutine_params(["server" => $param0, "frame" => $param1, "connection" => $conn]); - try { - (new MessageEvent($param0, $param1))->onActivate()->onAfter(); - } catch (Error $e) { - $error_msg = $e->getMessage() . " at " . $e->getFile() . "(" . $e->getLine() . ")"; - Console::error("Fatal error when calling $event_name: " . $error_msg); - Console::trace(); - } - break; - case "request": - try { - set_coroutine_params(["request" => $param0, "response" => $param1]); - (new RequestEvent($param0, $param1))->onActivate()->onAfter(); - } catch (Exception $e) { - /** @var Response $param1 */ - $param1->status(500); - Console::info($param0->server["remote_addr"] . ":" . $param0->server["remote_port"] . - " [" . $param1->getStatusCode() . "] " . $param0->server["request_uri"] - ); - if (!$param1->isEnd()) { - if (ZMConfig::get("global", "debug_mode")) - $param1->end("Internal server error: " . $e->getMessage()); - else - $param1->end("Internal server error."); - } - Console::error("Internal server exception (500), caused by " . get_class($e)); - Console::log($e->getTraceAsString(), "gray"); - } catch (Error $e) { - /** @var Response $param1 */ - $param1->status(500); - Console::info($param0->server["remote_addr"] . ":" . $param0->server["remote_port"] . - " [" . $param1->getStatusCode() . "] " . $param0->server["request_uri"] - ); - $doc = "Internal server error
"; - $error_msg = $e->getMessage() . " at " . $e->getFile() . "(" . $e->getLine() . ")"; - if (Console::getLevel() >= 4) $doc .= $error_msg; - if (!$param1->isEnd()) $param1->end($doc); - Console::error("Internal server error (500): " . $error_msg); - Console::log($e->getTraceAsString(), "gray"); - } - break; - case "open": - /** @var Request $param1 */ - set_coroutine_params(["server" => $param0, "request" => $param1, "fd" => $param1->fd]); - try { - (new WSOpenEvent($param0, $param1))->onActivate()->onAfter(); - } catch (Error $e) { - $error_msg = $e->getMessage() . " at " . $e->getFile() . "(" . $e->getLine() . ")"; - Console::error("Fatal error when calling $event_name: " . $error_msg); - Console::trace(); - } - break; - case "close": - set_coroutine_params(["server" => $param0, "fd" => $param1]); - try { - (new WSCloseEvent($param0, $param1))->onActivate()->onAfter(); - } catch (Error $e) { - $error_msg = $e->getMessage() . " at " . $e->getFile() . "(" . $e->getLine() . ")"; - Console::error("Fatal error when calling $event_name: " . $error_msg); - Console::trace(); - } - break; - } - //Console::info(Console::setColor("Event: " . $event_name . " 运行了 " . round(microtime(true) - $starttime, 5) . " 秒", "gold")); - } - - /** - * @param $event_data - * @param $conn_or_response - * @param int $level - * @return bool - * @throws AnnotationException - */ - public static function callCQEvent($event_data, $conn_or_response, int $level = 0) { - ctx()->setCache("level", $level); - if ($level >= 5) { - Console::warning("Recursive call reached " . $level . " times"); - Console::trace(); - return false; - } - $starttime = microtime(true); - switch ($event_data["post_type"]) { - case "message": - $event = new CQ\MessageEvent($event_data, $conn_or_response, $level); - if ($event->onBefore()) $event->onActivate(); - $event->onAfter(); - return $event->hasReply(); - break; - case "notice": - $event = new CQ\NoticeEvent($event_data, $conn_or_response, $level); - if ($event->onBefore()) $event->onActivate(); - $event->onAfter(); - return true; - case "request": - $event = new CQ\RequestEvent($event_data, $conn_or_response, $level); - if ($event->onBefore()) $event->onActivate(); - $event->onAfter(); - return true; - case "meta_event": - $event = new CQ\MetaEvent($event_data, $conn_or_response, $level); - if ($event->onBefore()) $event->onActivate(); - return true; - } - unset($starttime); - return false; - } - - /** - * @param $req - * @throws AnnotationException - */ - public static function callCQResponse($req) { - Console::debug("收到来自API连接的回复：" . json_encode($req, 128 | 256)); - $status = $req["status"]; - $retcode = $req["retcode"]; - $data = $req["data"]; - if (isset($req["echo"]) && LightCache::isset("sent_api_" . $req["echo"])) { - $origin = LightCache::get("sent_api_" . $req["echo"]); - $self_id = $origin["self_id"]; - $response = [ - "status" => $status, - "retcode" => $retcode, - "data" => $data, - "self_id" => $self_id, - "echo" => $req["echo"] - ]; - set_coroutine_params(["cq_response" => $response]); - if (isset(ZMBuf::$events[CQAPIResponse::class][$req["retcode"]])) { - list($c, $method) = ZMBuf::$events[CQAPIResponse::class][$req["retcode"]]; - $class = new $c(["data" => $origin["data"]]); - call_user_func_array([$class, $method], [$origin["data"], $req]); - } - $origin_ctx = ctx()->copy(); - ctx()->setCache("action", $origin["data"]["action"] ?? "unknown"); - ctx()->setData($origin["data"]); - foreach (ZMBuf::$events[CQAPISend::class] ?? [] as $k => $v) { - if (($v->action == "" || $v->action == ctx()->getCache("action")) && $v->with_result) { - $c = $v->class; - self::callWithMiddleware($c, $v->method, context()->copy(), [ctx()->getCache("action"), $origin["data"]["params"] ?? [], ctx()->getRobotId()]); - if (context()->getCache("block_continue") === true) break; - } - } - set_coroutine_params($origin_ctx); - if (($origin["func"] ?? null) !== null) { - call_user_func($origin["func"], $response, $origin["data"]); - } elseif (($origin["coroutine"] ?? false) !== false) { - $r = LightCache::get("sent_api_" . $req["echo"]); - $r["result"] = $response; - LightCache::set("sent_api_" . $req["echo"], $r); - Co::resume($origin['coroutine']); - } - LightCache::unset("sent_api_" . $req["echo"]); - } - } - - public static function callCQAPISend($reply, ?ConnectionObject $connection) { - $action = $reply["action"] ?? null; - if ($action === null) { - Console::warning("API 激活事件异常！"); - return; - } - if (ctx() === null) $content = []; - else $content = ctx()->copy(); - go(function () use ($action, $reply, $connection, $content) { - set_coroutine_params($content); - context()->setCache("action", $action); - context()->setCache("reply", $reply); - foreach (ZMBuf::$events[CQAPISend::class] ?? [] as $k => $v) { - if (($v->action == "" || $v->action == $action) && !$v->with_result) { - $c = $v->class; - self::callWithMiddleware($c, $v->method, context()->copy(), [$reply["action"], $reply["params"] ?? [], $connection->getOption('connect_id')]); - if (context()->getCache("block_continue") === true) break; - } - } - }); - } - - /** - * @param $c - * @param $method - * @param array $class_construct - * @param array $func_args - * @param null $after_call - * @return mixed|null - * @throws AnnotationException - * @throws Exception - */ - public static function callWithMiddleware($c, $method, array $class_construct, array $func_args, $after_call = null) { - $return_value = null; - $plain_class = is_object($c) ? get_class($c) : $c; - if (isset(ZMBuf::$events[MiddlewareInterface::class][$plain_class][$method])) { - $middlewares = ZMBuf::$events[MiddlewareInterface::class][$plain_class][$method]; - $before_result = true; - $r = []; - foreach ($middlewares as $k => $middleware) { - if (!isset(ZMBuf::$events[MiddlewareClass::class][$middleware])) throw new AnnotationException("Annotation parse error: Unknown MiddlewareClass named \"{$middleware}\"!"); - $middleware_obj = ZMBuf::$events[MiddlewareClass::class][$middleware]; - $before = $middleware_obj["class"]; - //var_dump($middleware_obj); - $r[$k] = new $before(); - $r[$k]->class = is_object($c) ? get_class($c) : $c; - $r[$k]->method = $method; - if (isset($middleware_obj["before"])) { - $rs = $middleware_obj["before"]; - $before_result = $r[$k]->$rs(...$func_args); - if ($before_result === false) break; - } - } - if ($before_result) { - try { - if (is_object($c)) $class = $c; - elseif ($class_construct == []) $class = ZMUtil::getModInstance($c); - else $class = new $c($class_construct); - $result = $class->$method(...$func_args); - if (is_callable($after_call)) - $return_value = $after_call($result); - } catch (Exception $e) { - for ($i = count($middlewares) - 1; $i >= 0; --$i) { - $middleware_obj = ZMBuf::$events[MiddlewareClass::class][$middlewares[$i]]; - if (!isset($middleware_obj["exceptions"])) continue; - foreach ($middleware_obj["exceptions"] as $name => $method) { - if ($e instanceof $name) { - $r[$i]->$method($e); - context()->setCache("block_continue", true); - } - } - if (context()->getCache("block_continue") === true) return $return_value; - } - throw $e; - } - } - for ($i = count($middlewares) - 1; $i >= 0; --$i) { - $middleware_obj = ZMBuf::$events[MiddlewareClass::class][$middlewares[$i]]; - if (isset($middleware_obj["after"], $r[$i])) { - $r[$i]->{$middleware_obj["after"]}(...$func_args); - } - } - } else { - if (is_object($c)) $class = $c; - elseif ($class_construct == []) $class = ZMUtil::getModInstance($c); - else $class = new $c($class_construct); - $result = call_user_func_array([$class, $method], $func_args); - if (is_callable($after_call)) - $return_value = call_user_func_array($after_call, [$result]); - } - return $return_value; - } -} diff --git a/src/ZM/Event/EventManager.php b/src/ZM/Event/EventManager.php index e6546abc..3fea9f3e 100644 --- a/src/ZM/Event/EventManager.php +++ b/src/ZM/Event/EventManager.php @@ -11,7 +11,7 @@ use ZM\Annotation\AnnotationBase; use ZM\Annotation\AnnotationParser; use ZM\Annotation\Swoole\OnTick; use ZM\Console\Console; -use ZM\Store\ZMBuf; +use ZM\Store\ZMAtomic; class EventManager { @@ -20,7 +20,7 @@ class EventManager public static $middlewares = []; public static $req_mapping = []; - public static function addEvent($event_name, AnnotationBase $event_obj) { + public static function addEvent($event_name, ?AnnotationBase $event_obj) { self::$events[$event_name][] = $event_obj; } @@ -43,7 +43,7 @@ class EventManager Console::debug("Added Middleware-based timer: " . $plain_class . " -> " . $vss->method); Timer::tick($vss->tick_ms, function () use ($vss, $dispatcher) { set_coroutine_params([]); - if (ZMBuf::atomic("stop_signal")->get() != 0) { + if (ZMAtomic::get("stop_signal")->get() != 0) { Timer::clearAll(); return; } diff --git a/src/ZM/Event/ServerEventHandler.php b/src/ZM/Event/ServerEventHandler.php index 4551fedc..a7ebcebd 100644 --- a/src/ZM/Event/ServerEventHandler.php +++ b/src/ZM/Event/ServerEventHandler.php @@ -17,7 +17,7 @@ use Swoole\Timer; use ZM\Annotation\AnnotationParser; use ZM\Annotation\Http\RequestMapping; use ZM\Annotation\Swoole\OnWorkerStart; -use ZM\Annotation\Swoole\SwooleEvent; +use ZM\Annotation\Swoole\OnSwooleEvent; use ZM\Config\ZMConfig; use ZM\ConnectionManager\ManagerGM; use ZM\Console\Console; @@ -33,6 +33,8 @@ use ZM\Exception\DbException; use ZM\Framework; use ZM\Http\Response; use ZM\Module\QQBot; +use ZM\Store\MySQL\SqlPoolStorage; +use ZM\Store\Redis\ZMRedisPool; use ZM\Store\ZMBuf; use ZM\Utils\DataProvider; use ZM\Utils\HttpUtil; @@ -132,9 +134,9 @@ class ServerEventHandler foreach ($server->connections as $v) { $server->close($v); } - if (ZMBuf::$sql_pool !== null) { - ZMBuf::$sql_pool->close(); - ZMBuf::$sql_pool = null; + if (SqlPoolStorage::$sql_pool !== null) { + SqlPoolStorage::$sql_pool->close(); + SqlPoolStorage::$sql_pool = null; } // 这里执行的是只需要执行一遍的代码，比如终端监听器和键盘监听器 @@ -175,7 +177,7 @@ class ServerEventHandler } } $sql = ZMConfig::get("global", "sql_config"); - ZMBuf::$sql_pool = new PDOPool((new PDOConfig()) + SqlPoolStorage::$sql_pool = new PDOPool((new PDOConfig()) ->withHost($sql["sql_host"]) ->withPort($sql["sql_port"]) // ->withUnixSocket('/tmp/mysql.sock') @@ -188,6 +190,13 @@ class ServerEventHandler DB::initTableList(); } + // 开箱即用的Redis + $redis = ZMConfig::get("global", "redis_config"); + if($redis !== null && $redis["host"] != "") { + if (!extension_loaded("redis")) Console::error("Can not find redis extension.\n"); + else ZMRedisPool::init($redis); + } + $this->loadAnnotations(); //加载composer资源、phar外置包、注解解析注册等 //echo json_encode(debug_backtrace(), 128|256); @@ -238,11 +247,11 @@ class ServerEventHandler * @param Frame $frame */ public function onMessage($server, Frame $frame) { - Console::debug("Calling Swoole \"message\" from fd=" . $frame->fd); + Console::debug("Calling Swoole \"message\" from fd=" . $frame->fd.": ".TermColor::ITALIC.$frame->data.TermColor::RESET); unset(Context::$context[Co::getCid()]); $conn = ManagerGM::get($frame->fd); set_coroutine_params(["server" => $server, "frame" => $frame, "connection" => $conn]); - $dispatcher = new EventDispatcher(SwooleEvent::class); + $dispatcher = new EventDispatcher(OnSwooleEvent::class); $dispatcher->setRuleFunction(function ($v) { if ($v->getRule() == '') { return strtolower($v->type) == 'message'; @@ -278,7 +287,7 @@ class ServerEventHandler Console::debug("Calling Swoole \"request\" event from fd=" . $request->fd); set_coroutine_params(["request" => $request, "response" => $response]); - $dis = new EventDispatcher(); + $dis = new EventDispatcher(OnSwooleEvent::class); $dis->setRuleFunction(function ($v) { if ($v->getRule() == '') { return strtolower($v->type) == 'request'; @@ -322,7 +331,7 @@ class ServerEventHandler else $response->end("Internal server error."); } - Console::error("Internal server exception (500), caused by " . get_class($e)); + Console::error("Internal server exception (500), caused by " . get_class($e).": ".$e->getMessage()); Console::log($e->getTraceAsString(), "gray"); } catch (Error $e) { $response->status(500); @@ -354,7 +363,8 @@ class ServerEventHandler ManagerGM::pushConnect($request->fd, $type_conn); $conn = ManagerGM::get($request->fd); set_coroutine_params(["server" => $server, "request" => $request, "connection" => $conn, "fd" => $request->fd]); - $dispatcher = new EventDispatcher(SwooleEvent::class); + $conn->setOption("connect_id", strval($request->header["x-self-id"]) ?? ""); + $dispatcher = new EventDispatcher(OnSwooleEvent::class); $dispatcher->setRuleFunction(function ($v) { if ($v->getRule() == '') { return strtolower($v->type) == 'open'; @@ -389,7 +399,7 @@ class ServerEventHandler if ($conn === null) return; Console::debug("Calling Swoole \"close\" event from fd=" . $fd); set_coroutine_params(["server" => $server, "connection" => $conn, "fd" => $fd]); - $dispatcher = new EventDispatcher(SwooleEvent::class); + $dispatcher = new EventDispatcher(OnSwooleEvent::class); $dispatcher->setRuleFunction(function ($v) { if ($v->getRule() == '') { return strtolower($v->type) == 'close'; @@ -421,9 +431,13 @@ class ServerEventHandler */ public function onPipeMessage(Server $server, $src_worker_id, $data) { //var_dump($data, $server->worker_id); - unset(Context::$context[Co::getCid()]); + //unset(Context::$context[Co::getCid()]); $data = json_decode($data, true); - switch ($data["action"]) { + switch ($data["action"] ?? '') { + case "resume_ws_message": + $obj = $data["data"]; + Co::resume($obj["coroutine"]); + break; case "stop": Console::verbose('正在清理 #' . $server->worker_id . ' 的计时器'); Timer::clearAll(); @@ -434,6 +448,9 @@ class ServerEventHandler case 'echo': Console::success('接收到来自 #' . $src_worker_id . ' 的消息'); break; + case 'send': + $server->sendMessage(json_encode(["action" => "echo"]), $data["target"]); + break; default: echo $data . PHP_EOL; } @@ -481,17 +498,17 @@ class ServerEventHandler } //加载插件 - $plugins = ZMConfig::get("global", "plugins") ?? []; + $plugins = ZMConfig::get("global", "modules") ?? []; if (!isset($plugins["qqbot"])) $plugins["qqbot"] = true; if ($plugins["qqbot"]) { - $obj = new SwooleEvent(); + $obj = new OnSwooleEvent(); $obj->class = QQBot::class; $obj->method = 'handle'; $obj->type = 'message'; $obj->level = 99999; $obj->rule = 'connectIsQQ()'; - EventManager::addEvent(SwooleEvent::class, $obj); + EventManager::addEvent(OnSwooleEvent::class, $obj); } //TODO: 编写加载外部插件的方式 diff --git a/src/ZM/Event/Swoole/MessageEvent.php b/src/ZM/Event/Swoole/MessageEvent.php deleted file mode 100644 index 55fde62f..00000000 --- a/src/ZM/Event/Swoole/MessageEvent.php +++ /dev/null @@ -1,97 +0,0 @@ -server = $server; - $this->frame = $frame; - } - - /** - * @inheritDoc - */ - public function onActivate() { - $conn = ManagerGM::get(context()->getFrame()->fd); - try { - if ($conn->getName() == "qq") { - $data = json_decode(context()->getFrame()->data, true); - if (isset($data["post_type"])) { - set_coroutine_params(["data" => $data, "connection" => $conn]); - ctx()->setCache("level", 0); - Console::debug("Calling CQ Event from fd=" . $conn->getFd()); - EventHandler::callCQEvent($data, ManagerGM::get(context()->getFrame()->fd), 0); - } else{ - set_coroutine_params(["connection" => $conn]); - EventHandler::callCQResponse($data); - } - } - foreach (ZMBuf::$events[SwooleEvent::class] ?? [] as $v) { - if (strtolower($v->type) == "message" && $this->parseSwooleRule($v)) { - $c = $v->class; - EventHandler::callWithMiddleware( - $c, - $v->method, - ["server" => $this->server, "frame" => $this->frame, "connection" => $conn], - [$conn] - ); - if (context()->getCache("block_continue") === true) break; - } - } - } catch (Exception $e) { - Console::warning("Websocket message event exception: " . (($cs = $e->getMessage()) == "" ? get_class($e) : $cs)); - Console::warning("In ". $e->getFile() . " at line ".$e->getLine()); - } - return $this; - } - - /** - * @inheritDoc - */ - public function onAfter() { - foreach (ZMBuf::$events[SwooleEventAfter::class] ?? [] as $v) { - if (strtolower($v->type) == "message" && $this->parseSwooleRule($v) === true) { - $c = $v->class; - $class = new $c(); - call_user_func_array([$class, $v->method], []); - if (context()->getCache("block_continue") === true) break; - } - } - return $this; - } - - private function parseSwooleRule($v) { - switch (explode(":", $v->rule)[0]) { - case "connectType": //websocket连接类型 - if ($v->callback instanceof Closure) return call_user_func($v->callback, ManagerGM::get($this->frame->fd)); - break; - case "dataEqual": //handle websocket message事件时才能用 - if ($v->callback instanceof Closure) return call_user_func($v->callback, $this->frame->data); - break; - } - return true; - } -} diff --git a/src/ZM/Event/Swoole/RequestEvent.php b/src/ZM/Event/Swoole/RequestEvent.php deleted file mode 100644 index f05f7a54..00000000 --- a/src/ZM/Event/Swoole/RequestEvent.php +++ /dev/null @@ -1,186 +0,0 @@ -request = $request; - $this->response = $response; - } - - /** - * @return $this|SwooleEvent - * @throws Exception - */ - public function onActivate() { - foreach (ZMConfig::get("global", "http_header") as $k => $v) { - $this->response->setHeader($k, $v); - } - $uri = $this->request->server["request_uri"]; - Console::verbose($this->request->server["remote_addr"] . " request " . $uri); - $uri = explode("/", $uri); - $uri = array_diff($uri, ["..", "", "."]); - $node = EventManager::$req_mapping; - $params = []; - while (true) { - $r = array_shift($uri); - if ($r === null) break; - if (($cnt = count($node["son"] ?? [])) == 1) { - if (isset($node["param_route"])) { - foreach ($node["son"] as $k => $v) { - if ($v["id"] == $node["param_route"]) { - $node = $v; - $params[mb_substr($v["name"], 1, -1)] = $r; - continue 2; - } - } - } elseif ($node["son"][0]["name"] == $r) { - $node = $node["son"][0]; - continue; - } - } elseif ($cnt >= 1) { - if (isset($node["param_route"])) { - foreach ($node["son"] as $k => $v) { - if ($v["id"] == $node["param_route"]) { - $node = $v; - $params[mb_substr($v["name"], 1, -1)] = $r; - continue 2; - } - } - } - foreach ($node["son"] as $k => $v) { - if ($v["name"] == $r) { - $node = $v; - continue 2; - } - } - } - - if (ZMConfig::get("global", "static_file_server")["status"]) { - $base_dir = ZMConfig::get("global", "static_file_server")["document_root"]; - $base_index = ZMConfig::get("global", "static_file_server")["document_index"]; - $uri = $this->request->server["request_uri"]; - $path = realpath($base_dir . urldecode($uri)); - if ($path !== false) { - if (is_dir($path)) $path = $path . '/'; - $work = realpath(DataProvider::getWorkingDir()) . '/'; - if (strpos($path, $work) !== 0) { - $this->responseStatus(403); - return $this; - } - if (is_dir($path)) { - foreach ($base_index as $vp) { - if (is_file($path . $vp)) { - Console::info("[200] " . $uri . " (static)"); - $exp = strtolower(pathinfo($path . $vp)['extension'] ?? "unknown"); - $this->response->setHeader("Content-Type", ZMConfig::get("file_header")[$exp] ?? "application/octet-stream"); - $this->response->end(file_get_contents($path . $vp)); - return $this; - } - } - } elseif (is_file($path)) { - Console::info("[200] " . $uri . " (static)"); - $exp = strtolower(pathinfo($path)['extension'] ?? "unknown"); - $this->response->setHeader("Content-Type", ZMConfig::get("file_header")[$exp] ?? "application/octet-stream"); - $this->response->end(file_get_contents($path)); - return $this; - } - } - } - $this->response->status(404); - $this->response->end(HttpUtil::getHttpCodePage(404)); - return $this; - } - context()->setCache("params", $params); - - if (in_array(strtoupper($this->request->server["request_method"]), $node["request_method"] ?? [])) { //判断目标方法在不在里面 - $c_name = $node["class"]; - EventHandler::callWithMiddleware( - $c_name, - $node["method"], - ["request" => $this->request, "response" => &$this->response, "params" => $params], - [$params], - function ($result) { - if (is_string($result) && !$this->response->isEnd()) $this->response->end($result); - if ($this->response->isEnd()) context()->setCache("block_continue", true); - } - ); - } - foreach (ZMBuf::$events[SwooleEvent::class] ?? [] as $v) { - if (strtolower($v->type) == "request" && $this->parseSwooleRule($v)) { - $c = $v->class; - EventHandler::callWithMiddleware($c, $v->method, ["request" => $this->request, "response" => $this->response], []); - if (context()->getCache("block_continue") === true) break; - } - } - - if (!$this->response->isEnd()) { - $this->response->status(404); - $this->response->end(HttpUtil::getHttpCodePage(404)); - } - return $this; - } - - /** - * @inheritDoc - */ - public function onAfter() { - foreach (ZMBuf::$events[SwooleEventAfter::class] ?? [] as $v) { - if (strtolower($v->type) == "request" && $this->parseSwooleRule($v)) { - $c = $v->class; - $class = new $c(["request" => $this->request, "response" => $this->response]); - call_user_func_array([$class, $v->method], []); - if ($class->block_continue) break; - } - } - return $this; - } - - private function responseStatus(int $int) { - $this->response->status($int); - $this->response->end(); - } - - private function parseSwooleRule($v) { - switch (explode(":", $v->rule)[0]) { - case "containsGet": - case "containsPost": - if ($v->callback instanceof Closure) return call_user_func($v->callback, $this->request); - break; - case "containsJson": - $content = $this->request->rawContent(); - $content = json_decode($content, true); - if ($content === null) return false; - if ($v->callback instanceof Closure) return call_user_func($v->callback, $content); - break; - } - return true; - } -} diff --git a/src/ZM/Event/Swoole/SwooleEventInterface.php b/src/ZM/Event/Swoole/SwooleEventInterface.php deleted file mode 100644 index 2918f3ff..00000000 --- a/src/ZM/Event/Swoole/SwooleEventInterface.php +++ /dev/null @@ -1,20 +0,0 @@ -server = $server; - $this->fd = $fd; - } - - /** - * @inheritDoc - * @throws AnnotationException - */ - public function onActivate() { - Console::debug("Websocket closed #{$this->fd}"); - set_coroutine_params(["server" => $this->server, "fd" => $this->fd, "connection" => ManagerGM::get($this->fd)]); - foreach(ZMBuf::$events[SwooleEvent::class] ?? [] as $v) { - if(strtolower($v->type) == "close" && $this->parseSwooleRule($v)) { - $c = $v->class; - EventHandler::callWithMiddleware($c, $v->method, ["server" => $this->server, "fd" => $this->fd], []); - if(context()->getCache("block_continue") === true) break; - } - } - ManagerGM::popConnect($this->fd); - return $this; - } - - /** - * @inheritDoc - * @throws AnnotationException - */ - public function onAfter() { - foreach (ZMBuf::$events[SwooleEventAfter::class] ?? [] as $v) { - if (strtolower($v->type) == "close" && $this->parseSwooleRule($v) === true) { - $c = $v->class; - EventHandler::callWithMiddleware($c, $v->method, ["server" => $this->server, "fd" => $this->fd], []); - if(context()->getCache("block_continue") === true) break; - } - } - return $this; - } - - private function parseSwooleRule($v) { - switch (explode(":", $v->rule)[0]) { - case "connectType": //websocket连接类型 - if ($v->callback instanceof Closure) return call_user_func($v->callback, ManagerGM::get($this->fd)); - break; - } - return true; - } -} diff --git a/src/ZM/Event/Swoole/WSOpenEvent.php b/src/ZM/Event/Swoole/WSOpenEvent.php deleted file mode 100644 index b0117435..00000000 --- a/src/ZM/Event/Swoole/WSOpenEvent.php +++ /dev/null @@ -1,106 +0,0 @@ -server = $server; - $this->request = $request; - } - - /** - * @inheritDoc - * @throws AnnotationException - */ - public function onActivate() { - $type = strtolower($this->request->get["type"] ?? $this->request->header["x-client-role"] ?? ""); - $type_conn = $this->getTypeClassName($type); - ManagerGM::pushConnect($this->request->fd, $type_conn); - if ($type_conn == "qq") { - ManagerGM::setName($this->request->fd, "qq"); - $qq = $this->request->get["qq"] ?? $this->request->header["x-self-id"] ?? ""; - $self_token = ZMConfig::get("global", "access_token") ?? ""; - if (isset($this->request->header["authorization"])) { - Console::debug($this->request->header["authorization"]); - } - $remote_token = $this->request->get["token"] ?? (isset($this->request->header["authorization"]) ? explode(" ", $this->request->header["authorization"])[1] : ""); - if ($qq != "" && ($self_token == $remote_token)) { - ManagerGM::setOption($this->request->fd, "connect_id", $qq); - $this->conn = ManagerGM::get($this->request->fd); - } else { - $this->conn = ManagerGM::get($this->request->fd); - Console::warning("connection of CQ has invalid QQ or token!"); - Console::debug("Remote token: " . $remote_token); - } - } else { - $this->conn = ManagerGM::get($this->request->fd); - } - set_coroutine_params(["server" => $this->server, "request" => $this->request, "connection" => $this->conn]); - foreach (ZMBuf::$events[SwooleEvent::class] ?? [] as $v) { - if (strtolower($v->type) == "open" && $this->parseSwooleRule($v) === true) { - $c = $v->class; - EventHandler::callWithMiddleware( - $c, - $v->method, - ["server" => $this->server, "request" => $this->request, "connection" => $this->conn], - [$this->conn] - ); - if (context()->getCache("block_continue") === true) break; - } - } - return $this; - } - - /** - * @inheritDoc - */ - public function onAfter() { - if (!$this->server->exists($this->conn->getFd())) return $this; - foreach (ZMBuf::$events[SwooleEventAfter::class] ?? [] as $v) { - if (strtolower($v->type) == "open" && $this->parseSwooleRule($v) === true) { - $class = new $v["class"](); - call_user_func_array([$class, $v["method"]], [$this->conn]); - if (context()->getCache("block_continue") === true) break; - } - } - return $this; - } - - private function parseSwooleRule($v) { - switch (explode(":", $v->rule)[0]) { - case "connectType": //websocket连接类型 - if ($v->callback instanceof Closure) return call_user_func($v->callback, $this->conn); - break; - } - return true; - } - - -} diff --git a/src/ZM/Exception/NotInitializedException.php b/src/ZM/Exception/NotInitializedException.php new file mode 100644 index 00000000..0f30c04f --- /dev/null +++ b/src/ZM/Exception/NotInitializedException.php @@ -0,0 +1,12 @@ + ZMConfig::get("global", "port"), "log_level" => Console::getLevel(), "version" => ZM_VERSION, - "config" => $args["env"] === null ? 'global.php' : $args["env"], - "working_dir" => DataProvider::getWorkingDir() + "config" => $args["env"] === null ? 'global.php' : $args["env"] ]; if (isset(ZMConfig::get("global", "swoole")["task_worker_num"])) { $out["task_worker_num"] = ZMConfig::get("global", "swoole")["task_worker_num"]; @@ -94,6 +92,7 @@ class Framework if (($num = ZMConfig::get("global", "swoole")["worker_num"] ?? swoole_cpu_num()) != 1) { $out["worker_num"] = $num; } + $out["working_dir"] = DataProvider::getWorkingDir(); Console::printProps($out, $tty_width); self::$server->set($this->server_set); @@ -112,12 +111,16 @@ class Framework $asd = get_included_files(); // 注册 Swoole Server 的事件 $this->registerServerEvents(); - LightCache::init(ZMConfig::get("global", "light_cache") ?? [ - "size" => 2048, - "max_strlen" => 4096, - "hash_conflict_proportion" => 0.6, - "persistence_path" => realpath(DataProvider::getDataFolder()."_cache.json") - ]); + $r = ZMConfig::get("global", "light_cache") ?? [ + "size" => 1024, + "max_strlen" => 8192, + "hash_conflict_proportion" => 0.6, + "persistence_path" => realpath(DataProvider::getDataFolder() . "_cache.json"), + "auto_save_interval" => 900 + ]; + LightCache::init($r); + LightCacheInside::init(); + SpinLock::init($r["size"]); self::$server->start(); } catch (Exception $e) { Console::error("Framework初始化出现错误，请检查！"); @@ -189,12 +192,12 @@ class Framework foreach ($args as $x => $y) { switch ($x) { case 'disable-coroutine': - if($y) { + if ($y) { $coroutine_mode = false; } break; case 'debug-mode': - if ($y) { + if ($y || ZMConfig::get("global", "debug_mode")) { $coroutine_mode = false; $terminal_id = null; Console::warning("You are in debug mode, do not use in production!"); @@ -243,11 +246,7 @@ class Framework if ($coroutine_mode) Runtime::enableCoroutine(true, SWOOLE_HOOK_ALL); } - private function getTtyWidth() { + public static function getTtyWidth() { return explode(" ", trim(exec("stty size")))[1]; } - - public static function getServer() { - return self::$server; - } } diff --git a/src/ZM/Module/QQBot.php b/src/ZM/Module/QQBot.php index 0de1be68..f1cd1de2 100644 --- a/src/ZM/Module/QQBot.php +++ b/src/ZM/Module/QQBot.php @@ -3,6 +3,23 @@ namespace ZM\Module; +use Swoole\Coroutine; +use ZM\Annotation\CQ\CQAPIResponse; +use ZM\Annotation\CQ\CQBefore; +use ZM\Annotation\CQ\CQCommand; +use ZM\Annotation\CQ\CQMessage; +use ZM\Annotation\CQ\CQMetaEvent; +use ZM\Annotation\CQ\CQNotice; +use ZM\Annotation\CQ\CQRequest; +use ZM\Console\Console; +use ZM\Console\TermColor; +use ZM\Event\EventDispatcher; +use ZM\Exception\InterruptException; +use ZM\Exception\WaitTimeoutException; +use ZM\Store\LightCacheInside; +use ZM\Store\Lock\SpinLock; +use ZM\Utils\CoMessage; + /** * Class QQBot * @package ZM\Module @@ -10,7 +27,170 @@ namespace ZM\Module; */ class QQBot { + /** + * @throws InterruptException + */ public function handle() { - //TODO: 写处理机器人事件的函数 + try { + $data = json_decode(context()->getFrame()->data, true); + if (isset($data["post_type"])) { + //echo TermColor::ITALIC.json_encode($data, 128|256).TermColor::RESET.PHP_EOL; + set_coroutine_params(["data" => $data]); + ctx()->setCache("level", 0); + //Console::debug("Calling CQ Event from fd=" . ctx()->getConnection()->getFd()); + $this->dispatchBeforeEvents($data); // >= 200 的level before在这里执行 + if (false) { + Console::error("哦豁，停下了"); + EventDispatcher::interrupt(); + } + //Console::warning("最上数据包：".json_encode($data)); + $this->dispatchEvents($data); + } else { + $this->dispatchAPIResponse($data); + } + } /** @noinspection PhpRedundantCatchClauseInspection */ catch (WaitTimeoutException $e) { + $e->module->finalReply($e->getMessage()); + } + } + + public function dispatchBeforeEvents($data) { + $before = new EventDispatcher(CQBefore::class); + $before->setRuleFunction(function ($v) use ($data) { + if ($v->level < 200) EventDispatcher::interrupt(); + elseif ($v->cq_event != $data["post_type"]) return false; + return true; + }); + $before->setReturnFunction(function ($result) { + if (!$result) EventDispatcher::interrupt(); + }); + $before->dispatchEvents($data); + } + + private function dispatchEvents($data) { + //Console::warning("最xia数据包：".json_encode($data)); + switch ($data["post_type"]) { + case "message": + $word = split_explode(" ", str_replace("\r", "", context()->getMessage())); + if (count(explode("\n", $word[0])) >= 2) { + $enter = explode("\n", context()->getMessage()); + $first = split_explode(" ", array_shift($enter)); + $word = array_merge($first, $enter); + foreach ($word as $k => $v) { + $word[$k] = trim($word[$k]); + } + } + + //分发CQCommand事件 + $dispatcher = new EventDispatcher(CQCommand::class); + $dispatcher->setRuleFunction(function ($v) use ($word) { + if ($v->match == "" && $v->regexMatch == "" && $v->fullMatch == "") return false; + elseif (($v->user_id == 0 || ($v->user_id != 0 && $v->user_id == ctx()->getUserId())) && + ($v->group_id == 0 || ($v->group_id != 0 && $v->group_id == (ctx()->getGroupId() ?? 0))) && + ($v->message_type == '' || ($v->message_type != '' && $v->message_type == ctx()->getMessageType())) + ) { + if (($word[0] != "" && $v->match == $word[0]) || + in_array($word[0], $v->alias) || + ($v->regexMatch != "" && ($args = matchArgs($v->regexMatch, ctx()->getMessage())) !== false) || + ($v->fullMatch != "" && (preg_match("/" . $v->fullMatch . "/u", ctx()->getMessage(), $args)) != 0)) { + return true; + } + } + return false; + }); + $dispatcher->setReturnFunction(function ($result) { + if (is_string($result)) ctx()->reply($result); + EventDispatcher::interrupt(); + }); + $r = $dispatcher->dispatchEvents($word); + if ($r === null) EventDispatcher::interrupt(); + + //分发CQMessage事件 + $msg_dispatcher = new EventDispatcher(CQMessage::class); + $msg_dispatcher->setRuleFunction(function ($v) { + return ($v->message == '' || ($v->message != '' && $v->message == context()->getData()["message"])) && + ($v->user_id == 0 || ($v->user_id != 0 && $v->user_id == context()->getData()["user_id"])) && + ($v->group_id == 0 || ($v->group_id != 0 && $v->group_id == (context()->getData()["group_id"] ?? 0))) && + ($v->message_type == '' || ($v->message_type != '' && $v->message_type == context()->getData()["message_type"])) && + ($v->raw_message == '' || ($v->raw_message != '' && $v->raw_message == context()->getData()["raw_message"])); + }); + $msg_dispatcher->setReturnFunction(function ($result) { + if (is_string($result)) ctx()->reply($result); + }); + $msg_dispatcher->dispatchEvents(ctx()->getMessage()); + return; + case "meta_event": + //Console::success("当前数据包：".json_encode(ctx()->getData())); + $dispatcher = new EventDispatcher(CQMetaEvent::class); + $dispatcher->setRuleFunction(function (CQMetaEvent $v) { + return ($v->meta_event_type == '' || ($v->meta_event_type != '' && $v->meta_event_type == ctx()->getData()["meta_event_type"])) && + ($v->sub_type == '' || ($v->sub_type != '' && $v->sub_type == (ctx()->getData()["sub_type"] ?? ''))); + }); + //eval(BP); + $dispatcher->dispatchEvents(ctx()->getData()); + return; + case "notice": + $dispatcher = new EventDispatcher(CQNotice::class); + $dispatcher->setRuleFunction(function (CQNotice $v) { + return + ($v->notice_type == '' || ($v->notice_type != '' && $v->notice_type == ctx()->getData()["notice_type"])) && + ($v->sub_type == '' || ($v->sub_type != '' && $v->sub_type == ctx()->getData()["sub_type"])) && + ($v->group_id == '' || ($v->group_id != '' && $v->group_id == ctx()->getData()["group_id"])) && + ($v->operator_id == '' || ($v->operator_id != '' && $v->operator_id == ctx()->getData()["operator_id"])); + }); + $dispatcher->dispatchEvents(ctx()->getData()); + return; + case "request": + $dispatcher = new EventDispatcher(CQRequest::class); + $dispatcher->setRuleFunction(function (CQRequest $v) { + return ($v->request_type == '' || ($v->request_type != '' && $v->request_type == ctx()->getData()['request_type'])) && + ($v->sub_type == '' || ($v->sub_type != '' && $v->sub_type == ctx()->getData()['sub_type'])) && + ($v->user_id == 0 || ($v->user_id != 0 && $v->user_id == ctx()->getData()["user_id"])) && + ($v->comment == '' || ($v->comment != '' && $v->comment == ctx()->getData()['comment'])); + }); + $dispatcher->dispatchEvents(ctx()->getData()); + return; + } + } + + private function dispatchAPIResponse($req) { + $status = $req["status"]; + $retcode = $req["retcode"]; + $data = $req["data"]; + if (isset($req["echo"]) && is_numeric($req["echo"])) { + $r = LightCacheInside::get("wait_api", "wait_api"); + if (isset($r[$req["echo"]])) { + $origin = $r[$req["echo"]]; + $self_id = $origin["self_id"]; + $response = [ + "status" => $status, + "retcode" => $retcode, + "data" => $data, + "self_id" => $self_id, + "echo" => $req["echo"] + ]; + set_coroutine_params(["cq_response" => $response]); + $dispatcher = new EventDispatcher(CQAPIResponse::class); + $dispatcher->setRuleFunction(function (CQAPIResponse $response) { + return $response->retcode == ctx()->getCQResponse()["retcode"]; + }); + $dispatcher->dispatchEvents($response); + + $origin_ctx = ctx()->copy(); + set_coroutine_params($origin_ctx); + if (($origin["coroutine"] ?? false) !== false) { + SpinLock::lock("wait_api"); + $r = LightCacheInside::get("wait_api", "wait_api"); + $r[$req["echo"]]["result"] = $response; + LightCacheInside::set("wait_api", "wait_api", $r); + SpinLock::unlock("wait_api"); + Coroutine::resume($origin['coroutine']); + } + SpinLock::lock("wait_api"); + $r = LightCacheInside::get("wait_api", "wait_api"); + unset($r[$req["echo"]]); + LightCacheInside::set("wait_api", "wait_api", $r); + SpinLock::unlock("wait_api"); + } + } } } diff --git a/src/ZM/Store/LightCache.php b/src/ZM/Store/LightCache.php index 964f8f71..5607c2e3 100644 --- a/src/ZM/Store/LightCache.php +++ b/src/ZM/Store/LightCache.php @@ -12,6 +12,8 @@ class LightCache { /** @var Table|null */ private static $kv_table = null; + /** @var Table|null */ + private static $kv_lock = null; private static $config = []; @@ -22,17 +24,21 @@ class LightCache self::$kv_table = new Table($config["size"], $config["hash_conflict_proportion"]); self::$kv_table->column("value", Table::TYPE_STRING, $config["max_strlen"]); self::$kv_table->column("expire", Table::TYPE_INT); - self::$kv_table->column("data_type", Table::TYPE_STRING, 12); + self::$kv_table->column("data_type", Table::TYPE_STRING, 8); $result = self::$kv_table->create(); + self::$kv_lock = new Table($config["size"], $config["hash_conflict_proportion"]); + $result = $result && self::$kv_lock->create(); if ($result === true && isset($config["persistence_path"])) { - $r = json_decode(file_get_contents($config["persistence_path"]), true); - if ($r === null) $r = []; - foreach ($r as $k => $v) { - $write = self::set($k, $v); - Console::debug("Writing LightCache: " . $k); - if ($write === false) { - self::$last_error = '可能是由于 Hash 冲突过多导致动态空间无法分配内存'; - return false; + if (file_exists($config["persistence_path"])) { + $r = json_decode(file_get_contents($config["persistence_path"]), true); + if ($r === null) $r = []; + foreach ($r as $k => $v) { + $write = self::set($k, $v, -2); + Console::verbose("Writing LightCache: " . $k); + if ($write === false) { + self::$last_error = '可能是由于 Hash 冲突过多导致动态空间无法分配内存'; + return false; + } } } } @@ -83,13 +89,46 @@ class LightCache $data_type = ""; } elseif (is_int($value)) { $data_type = "int"; - } else { + } elseif (is_bool($value)) { + $data_type = "bool"; + $value = json_encode($value); + }else { throw new Exception("Only can set string, array and int"); } try { return self::$kv_table->set($key, [ "value" => $value, - "expire" => $expire != -1 ? $expire + time() : -1, + "expire" => $expire >= 0 ? $expire + time() : $expire, + "data_type" => $data_type + ]); + } catch (Exception $e) { + return false; + } + } + + /** + * @param string $key + * @param $value + * @return bool|mixed + * @throws Exception + */ + public static function update(string $key, $value) { + if (self::$kv_table === null) throw new Exception("not initialized LightCache."); + if (is_array($value)) { + $value = json_encode($value, JSON_UNESCAPED_UNICODE); + if (strlen($value) >= self::$config["max_strlen"]) return false; + $data_type = "json"; + } elseif (is_string($value)) { + $data_type = ""; + } elseif (is_int($value)) { + $data_type = "int"; + } else { + throw new Exception("Only can set string, array and int"); + } + try { + if(self::$kv_table->get($key) === false) return false; + return self::$kv_table->set($key, [ + "value" => $value, "data_type" => $data_type ]); } catch (Exception $e) { @@ -131,14 +170,16 @@ class LightCache } public static function savePersistence() { + if(self::$kv_table === null) return; $r = []; foreach (self::$kv_table as $k => $v) { if ($v["expire"] === -2) { + Console::verbose("Saving " . $k); $r[$k] = self::parseGet($v); } } $r = file_put_contents(self::$config["persistence_path"], json_encode($r, 128 | 256)); - if($r === false) Console::error("Not saved, please check your \"persistence_path\"!"); + if ($r === false) Console::error("Not saved, please check your \"persistence_path\"!"); } private static function checkExpire($key) { @@ -153,6 +194,7 @@ class LightCache switch ($r["data_type"]) { case "json": case "int": + case "bool": return json_decode($r["value"], true); case "": default: diff --git a/src/ZM/Store/LightCacheInside.php b/src/ZM/Store/LightCacheInside.php new file mode 100644 index 00000000..f3838cc7 --- /dev/null +++ b/src/ZM/Store/LightCacheInside.php @@ -0,0 +1,55 @@ +column("value", Table::TYPE_STRING, 65536); + $result = self::$kv_table["wait_api"]->create(); + if ($result === false) { + self::$last_error = '系统内存不足，申请失败'; + return $result; + } + return $result; + } + + public static function get(string $table, string $key) { + if (!isset(self::$kv_table[$table])) throw new Exception("not initialized LightCache"); + $r = self::$kv_table[$table]->get($key); + return $r === false ? null : json_decode($r["value"], true); + } + + /** + * @param string $table + * @param string $key + * @param string|array|int $value + * @return mixed + * @throws Exception + */ + public static function set(string $table, string $key, $value) { + if (self::$kv_table === null) throw new Exception("not initialized LightCache"); + try { + return self::$kv_table[$table]->set($key, [ + "value" => json_encode($value, 256) + ]); + } catch (Exception $e) { + return false; + } + } + + public static function unset(string $table, string $key) { + return self::$kv_table[$table]->del($key); + } +} diff --git a/src/ZM/Store/Lock/SpinLock.php b/src/ZM/Store/Lock/SpinLock.php new file mode 100644 index 00000000..25b85fa4 --- /dev/null +++ b/src/ZM/Store/Lock/SpinLock.php @@ -0,0 +1,44 @@ +column('lock_num', Table::TYPE_INT, 8); + return self::$kv_lock->create(); + } + + public static function lock(string $key) + { + while (($r = self::$kv_lock->incr($key, 'lock_num')) > 1) { //此资源已经被锁上了 + if(Coroutine::getCid() != -1) System::sleep(self::$delay / 1000); + else usleep(self::$delay * 1000); + } + } + + public static function tryLock(string $key) { + if (($r = self::$kv_lock->incr($key, 'lock_num')) > 1) { + return false; + } + return true; + } + + public static function unlock(string $key) { + return self::$kv_lock->set($key, ['lock_num' => 0]); + } +} diff --git a/src/ZM/Store/MySQL/SqlPoolStorage.php b/src/ZM/Store/MySQL/SqlPoolStorage.php new file mode 100644 index 00000000..1df61eda --- /dev/null +++ b/src/ZM/Store/MySQL/SqlPoolStorage.php @@ -0,0 +1,13 @@ +get(); + $result = $callable($r); + if (isset($r->wasted)) ZMRedisPool::$pool->put(null); + else ZMRedisPool::$pool->put($r); + return $result; + } + + /** + * ZMRedis constructor. + * @throws NotInitializedException + */ + public function __construct() { + if(ZMRedisPool::$pool === null) throw new NotInitializedException("Redis pool is not initialized."); + $this->conn = ZMRedisPool::$pool->get(); + } + + /** + * @return Redis + */ + public function get() { + return $this->conn; + } + + public function __destruct() { + if (isset($this->conn->wasted)) ZMRedisPool::$pool->put(null); + else ZMRedisPool::$pool->put($this->conn); + } +} diff --git a/src/ZM/Store/Redis/ZMRedisPool.php b/src/ZM/Store/Redis/ZMRedisPool.php new file mode 100644 index 00000000..2ca1473e --- /dev/null +++ b/src/ZM/Store/Redis/ZMRedisPool.php @@ -0,0 +1,37 @@ +withHost($config['host']) + ->withPort($config['port']) + ->withAuth($config['auth']) + ->withDbIndex($config['db_index']) + ->withTimeout($config['timeout'] ?? 1) + ); + try { + $r = self::$pool->get()->ping('123'); + if(strpos(strtolower($r), "123") !== false) { + Console::debug("成功连接redis连接池！"); + } else { + var_dump($r); + } + } catch (RedisException $e) { + Console::error("Redis init failed! ".$e->getMessage()); + self::$pool = null; + } + } +} diff --git a/src/ZM/Store/ZMAtomic.php b/src/ZM/Store/ZMAtomic.php new file mode 100644 index 00000000..850a465e --- /dev/null +++ b/src/ZM/Store/ZMAtomic.php @@ -0,0 +1,38 @@ + $v) { + self::$atomics[$k] = new Atomic($v); + } + self::$atomics["stop_signal"] = new Atomic(0); + self::$atomics["wait_msg_id"] = new Atomic(0); + for ($i = 0; $i < 10; ++$i) { + self::$atomics["_tmp_" . $i] = new Atomic(0); + } + } + + +} diff --git a/src/ZM/Store/ZMBuf.php b/src/ZM/Store/ZMBuf.php index 25ef3a90..32c0621a 100755 --- a/src/ZM/Store/ZMBuf.php +++ b/src/ZM/Store/ZMBuf.php @@ -8,46 +8,10 @@ namespace ZM\Store; -use Swoole\Atomic; -use Swoole\Database\PDOPool; -use ZM\Config\ZMConfig; - class ZMBuf { - //读写的缓存数据，需要在worker_num = 1下才能正常使用 - //Swoole SQL连接池，多进程下每个进程一个连接池 - /** @var PDOPool */ - static $sql_pool = null;//保存sql连接池的类 - - /** @var array 事件注解的绑定对 */ public static $events = []; - - // 下面的有用，上面的没用了 - /** @var Atomic[] */ - public static $atomics; public static $instance = []; public static $context_class = []; public static $terminal = null; - - /** - * 初始化atomic计数器 - */ - public static function initAtomic() { - foreach (ZMConfig::get("global", "init_atomics") as $k => $v) { - self::$atomics[$k] = new Atomic($v); - } - self::$atomics["stop_signal"] = new Atomic(0); - self::$atomics["wait_msg_id"] = new Atomic(0); - for($i = 0; $i < 10; ++$i) { - self::$atomics["_tmp_".$i] = new Atomic(0); - } - } - - /** - * @param $name - * @return Atomic|null - */ - public static function atomic($name) { - return self::$atomics[$name] ?? null; - } } diff --git a/src/ZM/Utils/CoMessage.php b/src/ZM/Utils/CoMessage.php new file mode 100644 index 00000000..485229e0 --- /dev/null +++ b/src/ZM/Utils/CoMessage.php @@ -0,0 +1,80 @@ +add(1); + $hang["compare"] = $compare; + $hang["coroutine"] = $cid; + $hang["worker_id"] = server()->worker_id; + $hang["result"] = null; + SpinLock::lock("wait_api"); + $wait = LightCacheInside::get("wait_api", "wait_api"); + $wait[$api_id] = $hang; + LightCacheInside::set("wait_api", "wait_api", $wait); + SpinLock::unlock("wait_api"); + $id = swoole_timer_after($timeout * 1000, function () use ($api_id) { + $r = LightCacheInside::get("wait_api", "wait_api")[$api_id] ?? null; + if (is_array($r)) { + Co::resume($r["coroutine"]); + } + }); + Co::suspend(); + SpinLock::lock("wait_api"); + $sess = LightCacheInside::get("wait_api", "wait_api"); + $result = $sess[$api_id]["result"]; + unset($sess[$api_id]); + LightCacheInside::set("wait_api", "wait_api", $sess); + SpinLock::unlock("wait_api"); + if (isset($id)) swoole_timer_clear($id); + if ($result === null) return false; + return $result; + } + + public static function resumeByWS() { + $dat = ctx()->getData(); + $last = null; + SpinLock::lock("wait_api"); + $all = LightCacheInside::get("wait_api", "wait_api") ?? []; + foreach ($all as $k => $v) { + foreach ($v["compare"] as $vs) { + if ($v[$vs] != ($dat[$vs] ?? null)) { + continue 2; + } + } + $last = $k; + } + if($last !== null) { + $all[$last]["result"] = $dat; + LightCacheInside::set("wait_api", "wait_api", $all); + SpinLock::unlock("wait_api"); + if ($all[$last]["worker_id"] != server()->worker_id) { + ZMUtil::sendActionToWorker($all[$k]["worker_id"], "resume_ws_message", $all[$last]); + } else { + Co::resume($all[$last]["coroutine"]); + } + return true; + } else { + SpinLock::unlock("wait_api"); + return false; + } + } +} diff --git a/src/ZM/Utils/ZMUtil.php b/src/ZM/Utils/ZMUtil.php index 8b2e1960..928369fe 100644 --- a/src/ZM/Utils/ZMUtil.php +++ b/src/ZM/Utils/ZMUtil.php @@ -10,15 +10,18 @@ use Swoole\Event; use Swoole\Timer; use ZM\Console\Console; use ZM\Store\LightCache; +use ZM\Store\LightCacheInside; +use ZM\Store\ZMAtomic; use ZM\Store\ZMBuf; class ZMUtil { public static function stop() { Console::warning(Console::setColor("Stopping server...", "red")); + LightCache::savePersistence(); if (ZMBuf::$terminal !== null) Event::del(ZMBuf::$terminal); - ZMBuf::atomic("stop_signal")->set(1); + ZMAtomic::get("stop_signal")->set(1); try { LightCache::set('stop', 'OK'); } catch (Exception $e) { @@ -30,9 +33,8 @@ class ZMUtil public static function reload($delay = 800) { Console::info(Console::setColor("Reloading server...", "gold")); usleep($delay * 1000); - foreach (LightCache::getAll() as $k => $v) { - if (mb_substr($k, 0, 8) == "wait_api") - if ($v["result"] === null) Co::resume($v["coroutine"]); + foreach ((LightCacheInside::get("wait_api", "wait_api") ?? []) as $k => $v) { + if ($v["result"] === null && isset($v["coroutine"])) Co::resume($v["coroutine"]); } foreach (server()->connections as $v) { server()->close($v); @@ -50,4 +52,8 @@ class ZMUtil return ZMBuf::$instance[$class]; } } + + public static function sendActionToWorker($target_id, $action, $data) { + server()->sendMessage(json_encode(["action" => $action, "data" => $data]), $target_id); + } } diff --git a/src/ZM/Utils/terminal_listener.php b/src/ZM/Utils/terminal_listener.php deleted file mode 100644 index 24f72ac8..00000000 --- a/src/ZM/Utils/terminal_listener.php +++ /dev/null @@ -1,33 +0,0 @@ -set(['websocket_mask' => true]); - $client->setHeaders(["x-terminal-id" => $terminal_id, 'x-pid' => posix_getppid()]); - $ret = $client->upgrade("/?type=terminal"); - if ($ret) { - while (true) { - $line = fgets(STDIN); - if ($line !== false) { - $r = $client->push(trim($line)); - if (trim($line) == "reload" || trim($line) == "r" || trim($line) == "stop") { - break; - } - if($r === false) { - echo "Unable to connect framework terminal, connection closed. Trying to reconnect after 5s.\n"; - sleep(5); - goto hello; - } - } else { - break; - } - } - } else { - echo "Unable to connect framework terminal. port: $port\n"; - } -}); - diff --git a/src/ZM/global_defines.php b/src/ZM/global_defines.php index eaff3954..9612da44 100644 --- a/src/ZM/global_defines.php +++ b/src/ZM/global_defines.php @@ -5,10 +5,8 @@ use ZM\Config\ZMConfig; define("ZM_START_TIME", microtime(true)); define("ZM_DATA", ZMConfig::get("global", "zm_data")); define("ZM_VERSION", json_decode(file_get_contents(__DIR__ . "/../../composer.json"), true)["version"] ?? "unknown"); -define("CONFIG_DIR", ZMConfig::get("global", "config_dir")); define("CRASH_DIR", ZMConfig::get("global", "crash_dir")); @mkdir(ZM_DATA); -@mkdir(CONFIG_DIR); @mkdir(CRASH_DIR); define("CONN_WEBSOCKET", 0); diff --git a/src/ZM/global_functions.php b/src/ZM/global_functions.php index c0d462f5..605c74dd 100644 --- a/src/ZM/global_functions.php +++ b/src/ZM/global_functions.php @@ -1,5 +1,6 @@ 2048, + "size" => 2, "max_strlen" => 4096, - "hash_conflict_proportion" => 0.6, + "hash_conflict_proportion" => 0, + "persistence_path" => "../composer.json" ]); //LightCache::set("bool", true); $this->assertEquals(true, LightCache::set("2048", 123, 3)); $this->assertArrayHasKey("2048", LightCache::getAll()); sleep(3); $this->assertArrayNotHasKey("2048", LightCache::getAll()); + $this->assertEquals("Apache-2.0", LightCache::get("license")); + $this->assertEquals("zhamao/framework", LightCache::get("name")); + //$this->assertTrue(LightCache::set("storage", "asdasd", -2)); + //LightCache::savePersistence(); } } diff --git a/test/ZMTest/Mock/global.php b/test/ZMTest/Mock/global.php index f92f439c..c2f5915c 100644 --- a/test/ZMTest/Mock/global.php +++ b/test/ZMTest/Mock/global.php @@ -18,9 +18,6 @@ $config['debug_mode'] = false; /** 存放框架内文件数据的目录 */ $config['zm_data'] = realpath(__DIR__ . "/../") . '/zm_data/'; -/** 存放各个模块配置文件的目录 */ -$config['config_dir'] = $config['zm_data'] . 'config/'; - /** 存放崩溃和运行日志的目录 */ $config['crash_dir'] = $config['zm_data'] . 'crash/'; @@ -107,7 +104,7 @@ $config['command_register_class'] = [ ]; /** 服务器启用的外部第三方和内部插件 */ -$config['plugins'] = [ +$config['modules'] = [ 'qqbot' => true, // QQ机器人事件解析器，如果取消此项则默认为 true 开启状态，否则你手动填写 false 才会关闭 ];