as2252258/kiri-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

modify

Browse Source
This commit is contained in:
as2252258@163.com
2021-02-27 15:27:22 +08:00
parent 5ee9992afa
commit a3c6db9f5e
4 changed files with 477 additions and 457 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HttpServer/Events/OnWorkerStart.php
+15 -21
View File
@@ -27,6 +27,8 @@ class OnWorkerStart extends Callback
private int $_taskTable = 0; private int $_taskTable = 0;
private int $signal = SIGTERM | SIGKILL | SIGUSR2 | SIGUSR1;
/** /**
* @param Server $server * @param Server $server
@@ -38,12 +40,7 @@ class OnWorkerStart extends Callback
*/ */
public function onHandler(Server $server, int $worker_id): void public function onHandler(Server $server, int $worker_id): void
{ {
Coroutine::set([ Coroutine::set(['enable_deadlock_check' => false]);
'enable_deadlock_check' => false,
// 'exit_condition' => function () {
// return Coroutine::stats()['coroutine_num'] === 0;
// }
]);
putenv('workerId=' . $worker_id); putenv('workerId=' . $worker_id);
$get_name = $this->get_process_name($server, $worker_id); $get_name = $this->get_process_name($server, $worker_id);
@@ -52,11 +49,13 @@ class OnWorkerStart extends Callback
} }
$this->onSignal($server, $worker_id); $this->onSignal($server, $worker_id);
$this->debug(sprintf(workerName($worker_id) . ' #%d is start.....', $worker_id));
if ($worker_id >= $server->setting['worker_num']) { if ($worker_id >= $server->setting['worker_num']) {
fire(Event::SERVER_TASK_START); fire(Event::SERVER_TASK_START);
putenv('environmental=' . Snowflake::TASK);
} else { } else {
putenv('environmental=' . Snowflake::WORKER);
Snowflake::setWorkerId($server->worker_pid); Snowflake::setWorkerId($server->worker_pid);
$this->setWorkerAction($worker_id); $this->setWorkerAction($worker_id);
} }
@@ -68,22 +67,17 @@ class OnWorkerStart extends Callback
*/ */
public function onSignal($server, $worker_id) public function onSignal($server, $worker_id)
{ {
Coroutine\go(function (Server $server, $worker_id) { $this->debug(sprintf(workerName($worker_id) . ' #%d is start.....', $worker_id));
$sigkill = Coroutine::waitSignal(SIGTERM | SIGKILL | SIGUSR2 | SIGUSR1); Coroutine\go(function (Server $server) {
$sigkill = Coroutine::waitSignal($this->signal);
Timer::clearAll();
if ($sigkill === false) { if ($sigkill === false) {
do { return $server->stop();
$number = Co::stats()['coroutine_num']; }
var_dump($number); while (Co::stats()['coroutine_num'] > 0) {
if ($number === 0) { Coroutine::sleep(0.01);
break;
}
Coroutine::sleep(0.01);
} while (true);
} }
return $server->stop(); return $server->stop();
}, $server, $worker_id); }, $server);
} }
HttpServer/Server.php
+408 -412
View File
@@ -43,479 +43,475 @@ use Swoole\Runtime;
*/ */
class Server extends HttpService class Server extends HttpService
{ {
use Action; use Action;
const HTTP = 'HTTP'; const HTTP = 'HTTP';
const TCP = 'TCP'; const TCP = 'TCP';
const PACKAGE = 'PACKAGE'; const PACKAGE = 'PACKAGE';
const WEBSOCKET = 'WEBSOCKET'; const WEBSOCKET = 'WEBSOCKET';
private array $listening = []; private array $listening = [];
private array $server = [ private array $server = [
'HTTP' => [SWOOLE_TCP, Http::class], 'HTTP' => [SWOOLE_TCP, Http::class],
'TCP' => [SWOOLE_TCP, Receive::class], 'TCP' => [SWOOLE_TCP, Receive::class],
'PACKAGE' => [SWOOLE_UDP, Packet::class], 'PACKAGE' => [SWOOLE_UDP, Packet::class],
'WEBSOCKET' => [SWOOLE_SOCK_TCP, Websocket::class], 'WEBSOCKET' => [SWOOLE_SOCK_TCP, Websocket::class],
]; ];
private Packet|Websocket|Receive|null|Http $baseServer = null; private Packet|Websocket|Receive|null|Http $baseServer = null;
public int $daemon = 0; public int $daemon = 0;
private array $listenTypes = []; private array $listenTypes = [];
private array $process = [ private array $process = [
'biomonitoring' => Biomonitoring::class 'biomonitoring' => Biomonitoring::class
]; ];
private array $params = []; private array $params = [];
/** /**
* @param $name * @param $name
* @param $process * @param $process
* @param array $params * @param array $params
*/ */
public function addProcess($name, $process, $params = []) public function addProcess($name, $process, $params = [])
{ {
$this->process[$name] = $process; $this->process[$name] = $process;
$this->params[$name] = $params; $this->params[$name] = $params;
} }
/** /**
* @return array * @return array
*/ */
public function getProcesses(): array public function getProcesses(): array
{ {
return $this->process ?? []; return $this->process ?? [];
} }
/** /**
* @param array $configs * @param array $configs
* @return Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
public function initCore(array $configs): Packet|Websocket|Receive|Http|null public function initCore(array $configs): Packet|Websocket|Receive|Http|null
{ {
$this->enableCoroutine((bool)Config::get('settings.enable_coroutine')); $this->enableCoroutine((bool)Config::get('settings.enable_coroutine'));
$this->orders($configs); $this->orders($configs);
$this->onProcessListener(); $this->onProcessListener();
return $this->getServer(); return $this->getServer();
} }
/** /**
* @param $configs * @param $configs
* @return Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
* @throws Exception * @throws Exception
*/ */
private function orders($configs): Packet|Websocket|Receive|Http|null private function orders($configs): Packet|Websocket|Receive|Http|null
{ {
$servers = $this->sortServers($configs); $servers = $this->sortServers($configs);
foreach ($servers as $server) { foreach ($servers as $server) {
$this->create($server); $this->create($server);
if (!$this->baseServer) { if (!$this->baseServer) {
return null; return null;
} }
} }
return $this->baseServer; return $this->baseServer;
} }
/** /**
* @return string start server * @return string start server
* *
* start server * start server
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
public function start(): string public function start(): string
{ {
$configs = Config::get('servers', true); $configs = Config::get('servers', true);
Snowflake::clearWorkerId(); Snowflake::clearWorkerId();
$baseServer = $this->initCore($configs); $baseServer = $this->initCore($configs);
if (!$baseServer) { if (!$baseServer) {
return 'ok'; return 'ok';
} }
return $baseServer->start(); return $baseServer->start();
} }
/** /**
* @param $host * @param $host
* @param $Port * @param $Port
* @return Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
*/ */
public function error_stop($host, $Port): Packet|Websocket|Receive|Http|null public function error_stop($host, $Port): Packet|Websocket|Receive|Http|null
{ {
$this->error(sprintf('Port %s::%d is already.', $host, $Port)); $this->error(sprintf('Port %s::%d is already.', $host, $Port));
if ($this->baseServer) { if ($this->baseServer) {
$this->baseServer->shutdown(); $this->baseServer->shutdown();
} }
return $this->baseServer; return $this->baseServer;
} }
/** /**
* @return bool * @return bool
* @throws ConfigException * @throws ConfigException
*/ */
public function isRunner(): bool public function isRunner(): bool
{ {
$port = $this->sortServers(Config::get('servers')); $port = $this->sortServers(Config::get('servers'));
if (empty($port)) { if (empty($port)) {
return false; return false;
} }
if (Snowflake::isLinux()) { if (Snowflake::isLinux()) {
exec('netstat -tunlp | grep ' . $port[0]['port'], $output); exec('netstat -tunlp | grep ' . $port[0]['port'], $output);
} else { } else {
exec('lsof -i :' . $port[0]['port'] . ' | grep -i "LISTEN"', $output); exec('lsof -i :' . $port[0]['port'] . ' | grep -i "LISTEN"', $output);
} }
return !empty($output); return !empty($output);
} }
/** /**
* @return void * @return void
* *
* start server * start server
* @throws Exception * @throws Exception
*/ */
public function shutdown() public function shutdown()
{ {
$this->stop($this); $this->stop($this);
} }
/** /**
* @param bool $isEnable * @param bool $isEnable
*/ */
private function enableCoroutine($isEnable = true) private function enableCoroutine($isEnable = true)
{ {
if (!$isEnable) { if (!$isEnable) {
return; return;
} }
Runtime::enableCoroutine(true, SWOOLE_HOOK_TCP | Runtime::enableCoroutine(true, SWOOLE_HOOK_TCP |
SWOOLE_HOOK_UNIX | SWOOLE_HOOK_UNIX |
SWOOLE_HOOK_UDP | SWOOLE_HOOK_UDP |
SWOOLE_HOOK_UDG | SWOOLE_HOOK_UDG |
SWOOLE_HOOK_SSL | SWOOLE_HOOK_SSL |
SWOOLE_HOOK_TLS | SWOOLE_HOOK_TLS |
SWOOLE_HOOK_SLEEP | SWOOLE_HOOK_SLEEP |
SWOOLE_HOOK_STREAM_FUNCTION | SWOOLE_HOOK_STREAM_FUNCTION |
SWOOLE_HOOK_PROC SWOOLE_HOOK_PROC
); );
} }
/** /**
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
public function onProcessListener(): void public function onProcessListener(): void
{ {
if (!($this->baseServer instanceof \Swoole\Server)) { if (!($this->baseServer instanceof \Swoole\Server)) {
return; return;
} }
$processes = Config::get('processes'); $processes = Config::get('processes');
if (!empty($processes) && is_array($processes)) { if (!empty($processes) && is_array($processes)) {
$this->deliveryProcess(merge($processes, $this->process)); $this->deliveryProcess(merge($processes, $this->process));
} else { } else {
$this->deliveryProcess($this->process); $this->deliveryProcess($this->process);
} }
} }
/** /**
* @param $processes * @param $processes
* @throws Exception * @throws Exception
*/ */
private function deliveryProcess($processes) private function deliveryProcess($processes)
{ {
$application = Snowflake::app(); $application = Snowflake::app();
if (empty($processes) || !is_array($processes)) { if (empty($processes) || !is_array($processes)) {
return; return;
} }
foreach ($processes as $name => $process) { foreach ($processes as $name => $process) {
$is_enable_coroutine = true; $this->debug(sprintf('Process %s', $process));
if (is_array($process)) { if (!is_string($process)) {
[$process, $is_enable_coroutine] = $process; continue;
} }
$this->debug(sprintf('Process %s', $name . '::' . $process)); $system = new $process(Snowflake::app(), $name, true);
if (!is_string($process)) { if (isset($this->params[$name])) {
continue; $system->write(Json::encode($this->params[$name]));
} }
$system = new $process(Snowflake::app(), $name, $is_enable_coroutine); $this->baseServer->addProcess($system);
if (isset($this->params[$name])) { $application->set($process, $system);
$system->write(Json::encode($this->params[$name])); }
} }
$this->baseServer->addProcess($system);
$application->set($process, $system);
}
}
/** /**
* @param $daemon * @param $daemon
* @return Server * @return Server
*/ */
public function setDaemon($daemon): static public function setDaemon($daemon): static
{ {
if (!in_array($daemon, [0, 1])) { if (!in_array($daemon, [0, 1])) {
return $this; return $this;
} }
$this->daemon = $daemon; $this->daemon = $daemon;
return $this; return $this;
} }
/** /**
* @return Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
*/ */
public function getServer(): Packet|Websocket|Receive|Http|null public function getServer(): Packet|Websocket|Receive|Http|null
{ {
return $this->baseServer; return $this->baseServer;
} }
/** /**
* @param $config * @param $config
* @return mixed * @return mixed
* @throws Exception * @throws Exception
*/ */
private function create($config): mixed private function create($config): mixed
{ {
$settings = Config::get('settings', false, []); $settings = Config::get('settings', false, []);
if (!isset($this->server[$config['type']])) { if (!isset($this->server[$config['type']])) {
throw new Exception('Unknown server type(' . $config['type'] . ').'); throw new Exception('Unknown server type(' . $config['type'] . ').');
} }
$server = $this->dispatchCreate($config, $settings); $server = $this->dispatchCreate($config, $settings);
if (isset($config['events'])) { if (isset($config['events'])) {
$this->createEventListen($config); $this->createEventListen($config);
} }
return $server; return $server;
} }
/** /**
* @param $config * @param $config
* @throws Exception * @throws Exception
*/ */
protected function createEventListen($config) protected function createEventListen($config)
{ {
if (!is_array($config['events'])) { if (!is_array($config['events'])) {
return; return;
} }
$event = Snowflake::app()->getEvent(); $event = Snowflake::app()->getEvent();
foreach ($config['events'] as $name => $_event) { foreach ($config['events'] as $name => $_event) {
$event->on($name, $_event); $event->on($name, $_event);
} }
} }
/** /**
* @param $config * @param $config
* @param $settings * @param $settings
* @return \Swoole\Server|Packet|Receive|Http|Websocket|null * @return \Swoole\Server|Packet|Receive|Http|Websocket|null
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
* @throws Exception * @throws Exception
*/ */
private function dispatchCreate($config, $settings): \Swoole\Server|Packet|Receive|Http|Websocket|null private function dispatchCreate($config, $settings): \Swoole\Server|Packet|Receive|Http|Websocket|null
{ {
if (!($this->baseServer instanceof \Swoole\Server)) { if (!($this->baseServer instanceof \Swoole\Server)) {
$this->parseServer($config, $settings); $this->parseServer($config, $settings);
} else { } else {
if ($this->isUse($config['port'])) { if ($this->isUse($config['port'])) {
return $this->error_stop($config['host'], $config['port']); return $this->error_stop($config['host'], $config['port']);
} }
$newListener = $this->baseServer->addlistener($config['host'], $config['port'], $config['mode']); $newListener = $this->baseServer->addlistener($config['host'], $config['port'], $config['mode']);
if (isset($config['settings']) && is_array($config['settings'])) { if (isset($config['settings']) && is_array($config['settings'])) {
$newListener->set($config['settings']); $newListener->set($config['settings']);
} }
$this->onListenerBind($config, $this->baseServer); $this->onListenerBind($config, $this->baseServer);
} }
return $this->baseServer; return $this->baseServer;
} }
/** /**
* @param $config * @param $config
* @param $settings * @param $settings
* @return Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
* @throws Exception * @throws Exception
*/ */
private function parseServer($config, $settings): Packet|Websocket|Receive|Http|null private function parseServer($config, $settings): Packet|Websocket|Receive|Http|null
{ {
$class = $this->dispatch($config['type']); $class = $this->dispatch($config['type']);
if ($this->isUse($config['port'])) { if ($this->isUse($config['port'])) {
return $this->error_stop($config['host'], $config['port']); return $this->error_stop($config['host'], $config['port']);
} }
if (isset($config['settings']) && !empty($config['settings'])) { if (isset($config['settings']) && !empty($config['settings'])) {
$settings = array_merge($settings, $config['settings']); $settings = array_merge($settings, $config['settings']);
} }
$this->baseServer = new $class($config['host'], $config['port'], SWOOLE_PROCESS, $config['mode']); $this->baseServer = new $class($config['host'], $config['port'], SWOOLE_PROCESS, $config['mode']);
$settings['daemonize'] = $this->daemon; $settings['daemonize'] = $this->daemon;
if (!isset($settings['pid_file'])) { if (!isset($settings['pid_file'])) {
$settings['pid_file'] = APP_PATH . 'storage/server.pid'; $settings['pid_file'] = APP_PATH . 'storage/server.pid';
} }
if ($this->baseServer instanceof Websocket) { if ($this->baseServer instanceof Websocket) {
$this->onLoadWebsocketHandler(); $this->onLoadWebsocketHandler();
} else if ($this->baseServer instanceof Http) { } else if ($this->baseServer instanceof Http) {
$this->onLoadHttpHandler(); $this->onLoadHttpHandler();
} }
$this->baseServer->set($settings); $this->baseServer->set($settings);
return $this->baseServer; return $this->baseServer;
} }
/** /**
* @param $config * @param $config
* @param $newListener * @param $newListener
* @return void * @return void
* @throws ReflectionException * @throws ReflectionException
* @throws Exception * @throws Exception
* @throws NotFindClassException * @throws NotFindClassException
*/ */
private function onListenerBind($config, $newListener) private function onListenerBind($config, $newListener)
{ {
$this->debug(sprintf('Listener %s::%d -> %s', $config['host'], $config['port'], $config['mode'])); $this->debug(sprintf('Listener %s::%d -> %s', $config['host'], $config['port'], $config['mode']));
if ($config['type'] == self::WEBSOCKET) { if ($config['type'] == self::WEBSOCKET) {
throw new Exception('Base server must instanceof \Swoole\Websocket\Server::class.'); throw new Exception('Base server must instanceof \Swoole\Websocket\Server::class.');
} else if (!in_array($config['type'], [self::HTTP, self::TCP, self::PACKAGE])) { } else if (!in_array($config['type'], [self::HTTP, self::TCP, self::PACKAGE])) {
throw new Exception('Unknown server type(' . $config['type'] . ').'); throw new Exception('Unknown server type(' . $config['type'] . ').');
} }
if (in_array($config['type'], $this->listenTypes)) { if (in_array($config['type'], $this->listenTypes)) {
return; return;
} }
if ($config['type'] == self::HTTP) { if ($config['type'] == self::HTTP) {
if (in_array($config['type'], $this->listenTypes)) { if (in_array($config['type'], $this->listenTypes)) {
throw new Exception('Base server must instanceof \Swoole\Websocket\Server::class.'); throw new Exception('Base server must instanceof \Swoole\Websocket\Server::class.');
} }
$this->onBind($newListener, 'request', [Snowflake::createObject(OnRequest::class), 'onHandler']); $this->onBind($newListener, 'request', [Snowflake::createObject(OnRequest::class), 'onHandler']);
} else { } else {
$this->noHttp($newListener, $config); $this->noHttp($newListener, $config);
} }
array_push($this->listenTypes, $config['type']); array_push($this->listenTypes, $config['type']);
} }
/** /**
* @param $newListener * @param $newListener
* @param $config * @param $config
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
* @throws Exception * @throws Exception
*/ */
private function noHttp($newListener, $config) private function noHttp($newListener, $config)
{ {
$this->onBind($newListener, 'connect', [Snowflake::createObject(OnConnect::class), 'onHandler']); $this->onBind($newListener, 'connect', [Snowflake::createObject(OnConnect::class), 'onHandler']);
$this->onBind($newListener, 'close', [Snowflake::createObject(OnClose::class), 'onHandler']); $this->onBind($newListener, 'close', [Snowflake::createObject(OnClose::class), 'onHandler']);
if ($config['type'] == self::TCP) { if ($config['type'] == self::TCP) {
$this->onBind($newListener, 'receive', [$class = new OnReceive(), 'onHandler']); $this->onBind($newListener, 'receive', [$class = new OnReceive(), 'onHandler']);
} else { } else {
$this->onBind($newListener, 'packet', [$class = new OnPacket(), 'onHandler']); $this->onBind($newListener, 'packet', [$class = new OnPacket(), 'onHandler']);
} }
$class->pack = $config['resolve']['pack'] ?? null; $class->pack = $config['resolve']['pack'] ?? null;
$class->unpack = $config['resolve']['unpack'] ?? null; $class->unpack = $config['resolve']['unpack'] ?? null;
} }
/** /**
* @param $server * @param $server
* @param $name * @param $name
* @param $callback * @param $callback
* @throws Exception * @throws Exception
*/ */
private function onBind($server, $name, $callback) private function onBind($server, $name, $callback)
{ {
if (in_array($name, $this->listening)) { if (in_array($name, $this->listening)) {
return; return;
} }
if ($name === 'request') { if ($name === 'request') {
$this->onLoadHttpHandler(); $this->onLoadHttpHandler();
} }
array_push($this->listening, $name); array_push($this->listening, $name);
$server->on($name, $callback); $server->on($name, $callback);
} }
/** /**
* Load router handler * Load router handler
* @throws Exception * @throws Exception
*/ */
public function onLoadHttpHandler() public function onLoadHttpHandler()
{ {
$event = Snowflake::app()->getEvent(); $event = Snowflake::app()->getEvent();
$event->on(Event::SERVER_WORKER_START, function () { $event->on(Event::SERVER_WORKER_START, function () {
$router = Snowflake::app()->getRouter(); $router = Snowflake::app()->getRouter();
$router->loadRouterSetting(); $router->loadRouterSetting();
$attributes = Snowflake::app()->getAttributes(); $attributes = Snowflake::app()->getAttributes();
$attributes->read(CONTROLLER_PATH, 'App\Http\Controllers', 'controllers'); $attributes->read(CONTROLLER_PATH, 'App\Http\Controllers', 'controllers');
}); });
} }
/** /**
* @throws Exception * @throws Exception
*/ */
public function onLoadWebsocketHandler() public function onLoadWebsocketHandler()
{ {
$event = Snowflake::app()->getEvent(); $event = Snowflake::app()->getEvent();
$event->on(Event::SERVER_WORKER_START, function () { $event->on(Event::SERVER_WORKER_START, function () {
$attributes = Snowflake::app()->getAttributes(); $attributes = Snowflake::app()->getAttributes();
$attributes->read(SOCKET_PATH, 'App\Websocket', 'sockets'); $attributes->read(SOCKET_PATH, 'App\Websocket', 'sockets');
}); });
} }
/** /**
* @param $type * @param $type
* @return string * @return string
*/ */
private function dispatch($type): string private function dispatch($type): string
{ {
$default = [ $default = [
self::HTTP => Http::class, self::HTTP => Http::class,
self::WEBSOCKET => Websocket::class, self::WEBSOCKET => Websocket::class,
self::TCP => Receive::class, self::TCP => Receive::class,
self::PACKAGE => Packet::class self::PACKAGE => Packet::class
]; ];
return $default[$type] ?? Receive::class; return $default[$type] ?? Receive::class;
} }
/** /**
* @param $servers * @param $servers
* @return array * @return array
*/ */
private function sortServers($servers): array private function sortServers($servers): array
{ {
$array = []; $array = [];
foreach ($servers as $server) { foreach ($servers as $server) {
switch ($server['type']) { switch ($server['type']) {
case self::WEBSOCKET: case self::WEBSOCKET:
array_unshift($array, $server); array_unshift($array, $server);
break; break;
case self::HTTP: case self::HTTP:
case self::PACKAGE | self::TCP: case self::PACKAGE | self::TCP:
$array[] = $server; $array[] = $server;
break; break;
default: default:
$array[] = $server; $array[] = $server;
} }
} }
return $array; return $array;
} }
} }
System/Process/Process.php
+40 -24
View File
@@ -17,31 +17,47 @@ use Snowflake\Snowflake;
abstract class Process extends \Swoole\Process implements SProcess abstract class Process extends \Swoole\Process implements SProcess
{ {
/** @var Application $application */ /** @var Application $application */
protected Application $application; protected Application $application;
/** /**
* Process constructor. * Process constructor.
* @param $application * @param $application
* @param $name * @param $name
* @param bool $enable_coroutine * @param bool $enable_coroutine
* @throws Exception * @throws Exception
*/ */
public function __construct($application, $name, $enable_coroutine = true) public function __construct($application, $name, $enable_coroutine = true)
{ {
$class = get_called_class(); parent::__construct([$this, '_load'], true, 1, $enable_coroutine);
parent::__construct(function ($process) use ($name, $class) { $this->application = $application;
fire(Event::SERVER_WORKER_START); Snowflake::setWorkerId($this->pid);
putenv('workerId=Process.0'); }
if (Snowflake::isLinux()) {
$prefix = ucfirst(rtrim(Snowflake::app()->id, ':')); /**
$this->name($prefix . ': ' . $class); * @param Process $process
} * @throws \Snowflake\Exception\ComponentException
$this->onHandler($process); */
}, false, 1, $enable_coroutine); private function _load(Process $process)
$this->application = $application; {
Snowflake::setWorkerId($this->pid); putenv('environmental=' . Snowflake::PROCESS);
}
fire(Event::SERVER_WORKER_START);
if (Snowflake::isLinux()) {
$this->name($this->getPrefix());
}
$this->onHandler($process);
}
/**
* @return string
*/
private function getPrefix(): string
{
return ucfirst(rtrim(Snowflake::app()->id, ':') . ': ' . get_called_class());
}
} }
System/Snowflake.php
+14
View File
@@ -400,6 +400,20 @@ class Snowflake
private static array $_autoload = []; private static array $_autoload = [];
const PROCESS = 'process';
const TASK = 'task';
const WORKER = 'worker';
/**
* @return string|null
*/
public static function getEnvironmental(): ?string
{
return env('environmental');
}
/** /**
* @param $class * @param $class
* @param $file * @param $file