zhamao-robot/zhamao-framework
1
0
Fork 0
mirror of https://github.com/zhamao-robot/zhamao-framework.git synced 2026-07-11 02:35:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

tmp connect

Browse Source
This commit is contained in:
jerry
2021-03-22 07:44:11 +08:00
parent b6d1f724e9
commit 0ff4e52ed3
26 changed files with 812 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
src/ZM/Utils/ProcessManager.php
View File

@@ -4,14 +4,110 @@
namespace ZM\Utils;
use Co;
use ZM\Annotation\Swoole\OnPipeMessageEvent;
use ZM\Console\Console;
use ZM\Event\EventDispatcher;
use ZM\Store\LightCache;
use ZM\Store\LightCacheInside;
use ZM\Store\WorkerCache;
class ProcessManager
{
public static function runOnTask($param, $timeout = 0.5, $dst_worker_id = -1) {
return server()->taskwait([
"action" => "runMethod",
"class" => $param["class"],
"method" => $param["method"],
"params" => $param["params"]
], $timeout, $dst_worker_id);
public static function workerAction($src_worker_id, $data) {
$server = server();
switch ($data["action"] ?? '') {
case "eval":
eval($data["data"]);
break;
case "call_static":
call_user_func_array([$data["data"]["class"], $data["data"]["method"]], $data["data"]["params"]);
break;
case "save_persistence":
LightCache::savePersistence();
break;
case "resume_ws_message":
$obj = $data["data"];
Co::resume($obj["coroutine"]);
break;
case "getWorkerCache":
$r = WorkerCache::get($data["key"]);
$action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r];
$server->sendMessage(json_encode($action, 256), $src_worker_id);
break;
case "setWorkerCache":
$r = WorkerCache::set($data["key"], $data["value"]);
$action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r];
$server->sendMessage(json_encode($action, 256), $src_worker_id);
break;
case "unsetWorkerCache":
$r = WorkerCache::unset($data["key"]);
$action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r];
$server->sendMessage(json_encode($action, 256), $src_worker_id);
break;
case "hasKeyWorkerCache":
$r = WorkerCache::hasKey($data["key"], $data["subkey"]);
$action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r];
$server->sendMessage(json_encode($action, 256), $src_worker_id);
break;
case "asyncAddWorkerCache":
WorkerCache::add($data["key"], $data["value"], true);
break;
case "asyncSubWorkerCache":
WorkerCache::sub($data["key"], $data["value"], true);
break;
case "asyncSetWorkerCache":
WorkerCache::set($data["key"], $data["value"], true);
break;
case "asyncUnsetWorkerCache":
WorkerCache::unset($data["key"], true);
break;
case "addWorkerCache":
$r = WorkerCache::add($data["key"], $data["value"]);
$action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r];
$server->sendMessage(json_encode($action, 256), $src_worker_id);
break;
case "subWorkerCache":
$r = WorkerCache::sub($data["key"], $data["value"]);
$action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r];
$server->sendMessage(json_encode($action, 256), $src_worker_id);
break;
case "returnWorkerCache":
WorkerCache::$transfer[$data["cid"]] = $data["value"];
zm_resume($data["cid"]);
break;
default:
$dispatcher = new EventDispatcher(OnPipeMessageEvent::class);
$dispatcher->setRuleFunction(function (OnPipeMessageEvent $v) use ($data) {
return $v->action == $data["action"];
});
$dispatcher->dispatchEvents($data);
break;
}
}
public static function sendActionToWorker($worker_id, $action, $data) {
$obj = ["action" => $action, "data" => $data];
if (server()->worker_id === -1 && server()->getManagerPid() != posix_getpid()) {
Console::warning("Cannot send worker action from master or manager process!");
return;
}
if (server()->worker_id == $worker_id) {
self::workerAction($worker_id, $obj);
} else {
server()->sendMessage(json_encode($obj), $worker_id);
}
}
public static function resumeAllWorkerCoroutines() {
if (server()->worker_id === -1) {
Console::warning("Cannot call '".__FUNCTION__."' in non-worker process!");
return;
}
foreach ((LightCacheInside::get("wait_api", "wait_api") ?? []) as $k => $v) {
if (($v["result"] ?? false) === null && isset($v["coroutine"], $v["worker_id"])) {
if (server()->worker_id == $v["worker_id"]) Co::resume($v["coroutine"]);
}
}
}
}