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-07-03 21:12:21 +08:00
parent 4c2068482a
commit 6612040b3a
2 changed files with 387 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HttpServer/Events/OnClose.php
+1 -2
View File
@@ -27,9 +27,8 @@ class OnClose extends Callback
{ {
try { try {
defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES)); defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES));
$clientInfo = $server->getClientInfo($fd,null, true);
Event::trigger($this->getName($clientInfo, Event::SERVER_CLIENT_CLOSE), [$server, $fd]); Event::trigger(Event::SERVER_CLIENT_CLOSE, [$server, $fd]);
} catch (\Throwable $exception) { } catch (\Throwable $exception) {
$this->addError($exception, 'throwable'); $this->addError($exception, 'throwable');
} }
HttpServer/Server.php
+386 -382
View File
@@ -33,87 +33,87 @@ defined('PID_PATH') or define('PID_PATH', APP_PATH . 'storage/server.pid');
class Server extends HttpService class Server extends HttpService
{ {
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 $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 $swoole = null; private Packet|Websocket|Receive|null|Http $swoole = null;
public int $daemon = 0; public int $daemon = 0;
private array $process = [ private array $process = [
'biomonitoring' => Biomonitoring::class, 'biomonitoring' => Biomonitoring::class,
'logger_process' => LoggerProcess::class 'logger_process' => LoggerProcess::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 $configs * @param $configs
* @return Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
* @throws Exception * @throws Exception
*/ */
private function initCore($configs): Packet|Websocket|Receive|Http|null private function initCore($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->swoole) { if (!$this->swoole) {
throw new Exception('Base service create fail.'); throw new Exception('Base service create fail.');
} }
} }
return $this->startRpcService(); return $this->startRpcService();
} }
/** /**
* @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);
$baseServer = $this->initCore($configs); $baseServer = $this->initCore($configs);
if (!$baseServer) { if (!$baseServer) {
return 'ok'; return 'ok';
} }
Runtime::enableCoroutine(true, SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_BLOCKING_FUNCTION); Runtime::enableCoroutine(true, SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_BLOCKING_FUNCTION);
$settings['enable_deadlock_check'] = false; $settings['enable_deadlock_check'] = false;
$settings['exit_condition'] = function () { $settings['exit_condition'] = function () {
@@ -121,368 +121,372 @@ class Server extends HttpService
}; };
Coroutine::set($settings); Coroutine::set($settings);
return $this->execute($baseServer); return $this->execute($baseServer);
} }
/** /**
* @param $baseServer * @param $baseServer
* @return mixed * @return mixed
* @throws Exception * @throws Exception
*/ */
private function execute($baseServer): mixed private function execute($baseServer): mixed
{ {
$app = Snowflake::app(); $app = Snowflake::app();
$app->set('base-server', $baseServer); $app->set('base-server', $baseServer);
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
* @throws Exception * @throws Exception
*/ */
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->swoole) { if ($this->swoole) {
$this->swoole->shutdown(); $this->swoole->shutdown();
} else { } else {
$this->shutdown(); $this->shutdown();
} }
return $this->swoole; return $this->swoole;
} }
/** /**
* @return bool * @return bool
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
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;
} }
foreach ($port as $value) { foreach ($port as $value) {
if ($this->checkPort($value['port'])) { if ($this->checkPort($value['port'])) {
return true; return true;
} }
} }
return false; return false;
} }
/** /**
* @param $port * @param $port
* @return bool * @return bool
* @throws Exception * @throws Exception
*/ */
private function checkPort($port): bool private function checkPort($port): bool
{ {
if (Snowflake::getPlatform()->isLinux()) { if (Snowflake::getPlatform()->isLinux()) {
exec('netstat -tunlp | grep ' . $port, $output); exec('netstat -tunlp | grep ' . $port, $output);
} else { } else {
exec('lsof -i :' . $port . ' | grep -i "LISTEN"', $output); exec('lsof -i :' . $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()
{ {
/** @var Shutdown $shutdown */ /** @var Shutdown $shutdown */
$shutdown = Snowflake::app()->get('shutdown'); $shutdown = Snowflake::app()->get('shutdown');
$shutdown->shutdown(); $shutdown->shutdown();
} }
/** /**
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
public function onProcessListener(): \Swoole\Server|null|Packet|Receive|Http|Websocket public function onProcessListener(): \Swoole\Server|null|Packet|Receive|Http|Websocket
{ {
if (!($this->swoole instanceof \Swoole\Server)) { if (!($this->swoole instanceof \Swoole\Server)) {
return $this->swoole; return $this->swoole;
} }
$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);
} }
return $this->swoole; return $this->swoole;
} }
/** /**
* @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) {
$this->debug(sprintf('Process %s', $process)); $this->debug(sprintf('Process %s', $process));
if (!is_string($process)) { if (!is_string($process)) {
continue; continue;
} }
$system = Snowflake::createObject($process, [Snowflake::app(), $name, true]); $system = Snowflake::createObject($process, [Snowflake::app(), $name, true]);
if (isset($this->params[$name]) && !empty($this->params[$name])) { if (isset($this->params[$name]) && !empty($this->params[$name])) {
$system->write(swoole_serialize($this->params[$name])); $system->write(swoole_serialize($this->params[$name]));
} }
$this->swoole->addProcess($system); $this->swoole->addProcess($system);
$application->set($process, $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->swoole; return $this->swoole;
} }
/** /**
* @param $config * @param $config
* @return \Swoole\Server|Packet|Receive|Http|Websocket|null * @return \Swoole\Server|Packet|Receive|Http|Websocket|null
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
private function create($config): \Swoole\Server|null|Packet|Receive|Http|Websocket private function create($config): \Swoole\Server|null|Packet|Receive|Http|Websocket
{ {
$settings = Config::get('settings', []); $settings = Config::get('settings', []);
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;
} }
foreach ($config['events'] as $name => $_event) { foreach ($config['events'] as $name => $_event) {
Event::on('listen ' . $config['port'] . ' ' . $name, $_event); if ($name === Event::SERVER_CLIENT_CLOSE) {
} Event::on($name, $_event);
} }else{
Event::on('listen ' . $config['port'] . ' ' . $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 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 (Snowflake::port_already($config['port'])) { if (Snowflake::port_already($config['port'])) {
return $this->error_stop($config['host'], $config['port']); return $this->error_stop($config['host'], $config['port']);
} }
if (!($this->swoole instanceof \Swoole\Server)) { if (!($this->swoole instanceof \Swoole\Server)) {
return $this->parseServer($config, $settings); return $this->parseServer($config, $settings);
} }
return $this->addListener($config); return $this->addListener($config);
} }
/** /**
* @param $config * @param $config
* @return Http|Packet|Receive|Websocket|null * @return Http|Packet|Receive|Websocket|null
* @throws Exception * @throws Exception
*/ */
private function addListener($config): Packet|Websocket|Receive|Http|null private function addListener($config): Packet|Websocket|Receive|Http|null
{ {
$newListener = $this->swoole->addlistener($config['host'], $config['port'], $config['mode']); $newListener = $this->swoole->addlistener($config['host'], $config['port'], $config['mode']);
if (!$newListener) { if (!$newListener) {
exit($this->addError(sprintf('Listen %s::%d fail.', $config['host'], $config['port']))); exit($this->addError(sprintf('Listen %s::%d fail.', $config['host'], $config['port'])));
} }
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->onListenerBind($config);
return $this->swoole; return $this->swoole;
} }
/** /**
* @return Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
* @throws ConfigException * @throws ConfigException
* @throws Exception * @throws Exception
*/ */
private function startRpcService(): Packet|Websocket|Receive|Http|null private function startRpcService(): Packet|Websocket|Receive|Http|null
{ {
$rpcService = Config::get('rpc', []); $rpcService = Config::get('rpc', []);
if (empty($rpcService)) { if (empty($rpcService)) {
return $this->swoole; return $this->swoole;
} }
$this->addListener($rpcService); $this->addListener($rpcService);
return $this->swoole; return $this->swoole;
} }
/** /**
* @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 (is_array($config['settings'] ?? null)) { if (is_array($config['settings'] ?? null)) {
$settings = array_merge($settings, $config['settings']); $settings = array_merge($settings, $config['settings']);
} }
$this->debug(Snowflake::listen($config)); $this->debug(Snowflake::listen($config));
$this->swoole = $this->createServer($class, $config); $this->swoole = $this->createServer($class, $config);
$this->swoole->set(array_merge($settings, [ $this->swoole->set(array_merge($settings, [
'daemonize' => $this->daemon, 'daemonize' => $this->daemon,
'pid_file' => $settings['pid_file'] ?? PID_PATH 'pid_file' => $settings['pid_file'] ?? PID_PATH
])); ]));
return $this->onProcessListener(); return $this->onProcessListener();
} }
/** /**
* @param $class * @param $class
* @param $config * @param $config
* @return mixed * @return mixed
*/ */
private function createServer($class, $config): mixed private function createServer($class, $config): mixed
{ {
return new $class($config['host'], $config['port'], SWOOLE_PROCESS, $config['mode']); return new $class($config['host'], $config['port'], SWOOLE_PROCESS, $config['mode']);
} }
/** /**
* @param $config * @param $config
* @return Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
* @throws Exception * @throws Exception
*/ */
private function onListenerBind($config): Packet|Websocket|Receive|Http|null private function onListenerBind($config): Packet|Websocket|Receive|Http|null
{ {
$this->bindServerEvent($config['type']); $this->bindServerEvent($config['type']);
$this->debug(sprintf('Check listen %s::%d -> ok', $config['host'], $config['port'])); $this->debug(sprintf('Check listen %s::%d -> ok', $config['host'], $config['port']));
return $this->swoole; return $this->swoole;
} }
/** /**
* @param string $type * @param string $type
* @throws Exception * @throws Exception
*/ */
private function bindServerEvent($type = self::TCP) private function bindServerEvent($type = self::TCP)
{ {
if (in_array($type, [self::PACKAGE, self::TCP])) { if (in_array($type, [self::PACKAGE, self::TCP])) {
$this->onBindCallback('connect', [make(OnConnect::class), 'onHandler']); $this->onBindCallback('connect', [make(OnConnect::class), 'onHandler']);
$this->onBindCallback('close', [make(OnClose::class), 'onHandler']); $this->onBindCallback('close', [make(OnClose::class), 'onHandler']);
if ($type == self::PACKAGE) { if ($type == self::PACKAGE) {
$this->onBindCallback('packet', [make(OnPacket::class), 'onHandler']); $this->onBindCallback('packet', [make(OnPacket::class), 'onHandler']);
} else if ($type == self::TCP) { } else if ($type == self::TCP) {
$this->onBindCallback('receive', [make(OnReceive::class), 'onHandler']); $this->onBindCallback('receive', [make(OnReceive::class), 'onHandler']);
} }
} else if ($type === self::HTTP) { } else if ($type === self::HTTP) {
$this->onBindCallback('request', [make(OnRequest::class), 'onHandler']); $this->onBindCallback('request', [make(OnRequest::class), 'onHandler']);
} else { } else {
throw new Exception('Unknown server type(' . $type . ').'); throw new Exception('Unknown server type(' . $type . ').');
} }
} }
/** /**
* @param $name * @param $name
* @param $callback * @param $callback
* @throws Exception * @throws Exception
*/ */
public function onBindCallback($name, $callback) public function onBindCallback($name, $callback)
{ {
if ($this->swoole->getCallback($name) !== null) { if ($this->swoole->getCallback($name) !== null) {
return; return;
} }
$this->swoole->on($name, $callback); $this->swoole->on($name, $callback);
} }
/** /**
* @param $type * @param $type
* @return string * @return string
*/ */
private function dispatch($type): string private function dispatch($type): string
{ {
return match ($type) { return match ($type) {
self::HTTP => Http::class, self::HTTP => Http::class,
self::WEBSOCKET => Websocket::class, self::WEBSOCKET => Websocket::class,
self::PACKAGE => Packet::class, self::PACKAGE => Packet::class,
default => Receive::class default => 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;
} }
} }