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-03-26 01:05:07 +08:00
parent d38c66a92d
commit a341734a33
1 changed files with 494 additions and 489 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HttpServer/Server.php
+494 -489
View File
@@ -22,6 +22,8 @@ use Rpc\Producer;
use Rpc\Service; use Rpc\Service;
use Snowflake\Abstracts\Config; use Snowflake\Abstracts\Config;
use Snowflake\Core\Json; use Snowflake\Core\Json;
use Snowflake\Crontab\Consumer;
use Snowflake\Crontab\CrontabZookeeperProcess;
use Snowflake\Error\LoggerProcess; use Snowflake\Error\LoggerProcess;
use Snowflake\Event; use Snowflake\Event;
use Snowflake\Exception\ComponentException; use Snowflake\Exception\ComponentException;
@@ -51,495 +53,498 @@ defined('PID_PATH') or define('PID_PATH', APP_PATH . 'storage/server.pid');
*/ */
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,
'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->baseServer) { if (!$this->baseServer) {
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);
if (Config::get('crontab.enable') === true) {
$baseServer = $this->initCore($configs); $this->addProcess('CrontabZookeeper', CrontabZookeeperProcess::class);
if (!$baseServer) { $this->addProcess('CrontabConsumer', Consumer::class);
return 'ok'; }
} $baseServer = $this->initCore($configs);
if (!$baseServer) {
Runtime::enableCoroutine(true, SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_BLOCKING_FUNCTION); return 'ok';
}
Coroutine::set(['enable_deadlock_check' => false]);
Runtime::enableCoroutine(true, SWOOLE_HOOK_ALL ^ SWOOLE_HOOK_BLOCKING_FUNCTION);
return $baseServer->start();
} Coroutine::set(['enable_deadlock_check' => false]);
return $baseServer->start();
/** }
* @param $host
* @param $Port
* @return Packet|Websocket|Receive|Http|null /**
* @throws Exception * @param $host
*/ * @param $Port
public function error_stop($host, $Port): Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
{ * @throws Exception
$this->error(sprintf('Port %s::%d is already.', $host, $Port)); */
if ($this->baseServer) { public function error_stop($host, $Port): Packet|Websocket|Receive|Http|null
$this->baseServer->shutdown(); {
} else { $this->error(sprintf('Port %s::%d is already.', $host, $Port));
$this->shutdown(); if ($this->baseServer) {
} $this->baseServer->shutdown();
return $this->baseServer; } else {
} $this->shutdown();
}
return $this->baseServer;
/** }
* @return bool
* @throws ConfigException
* @throws Exception /**
*/ * @return bool
public function isRunner(): bool * @throws ConfigException
{ * @throws Exception
$port = $this->sortServers(Config::get('servers')); */
if (empty($port)) { public function isRunner(): bool
return false; {
} $port = $this->sortServers(Config::get('servers'));
foreach ($port as $value) { if (empty($port)) {
if ($this->checkPort($value['port'])) { return false;
return true; }
} foreach ($port as $value) {
} if ($this->checkPort($value['port'])) {
return false; return true;
} }
}
return false;
/** }
* @param $port
* @return bool
* @throws Exception /**
*/ * @param $port
private function checkPort($port): bool * @return bool
{ * @throws Exception
if (Snowflake::getPlatform()->isLinux()) { */
exec('netstat -tunlp | grep ' . $port, $output); private function checkPort($port): bool
} else { {
exec('lsof -i :' . $port . ' | grep -i "LISTEN"', $output); if (Snowflake::getPlatform()->isLinux()) {
} exec('netstat -tunlp | grep ' . $port, $output);
return !empty($output); } else {
} exec('lsof -i :' . $port . ' | grep -i "LISTEN"', $output);
}
return !empty($output);
/** }
* @return void
*
* start server /**
* @throws Exception * @return void
*/ *
public function shutdown() * start server
{ * @throws Exception
$this->stop($this); */
} public function shutdown()
{
$this->stop($this);
/** }
* @throws ConfigException
* @throws Exception
*/ /**
public function onProcessListener(): void * @throws ConfigException
{ * @throws Exception
if (!($this->baseServer instanceof \Swoole\Server)) { */
return; public function onProcessListener(): void
} {
if (!($this->baseServer instanceof \Swoole\Server)) {
$processes = Config::get('processes'); return;
if (!empty($processes) && is_array($processes)) { }
$this->deliveryProcess(merge($processes, $this->process));
} else { $processes = Config::get('processes');
$this->deliveryProcess($this->process); if (!empty($processes) && is_array($processes)) {
} $this->deliveryProcess(merge($processes, $this->process));
} } else {
$this->deliveryProcess($this->process);
}
/** }
* @param $processes
* @throws Exception
*/ /**
private function deliveryProcess($processes) * @param $processes
{ * @throws Exception
$application = Snowflake::app(); */
if (empty($processes) || !is_array($processes)) { private function deliveryProcess($processes)
return; {
} $application = Snowflake::app();
foreach ($processes as $name => $process) { if (empty($processes) || !is_array($processes)) {
$this->debug(sprintf('Process %s', $process)); return;
if (!is_string($process)) { }
continue; foreach ($processes as $name => $process) {
} $this->debug(sprintf('Process %s', $process));
$system = Snowflake::createObject($process, [Snowflake::app(), $name, true]); if (!is_string($process)) {
if (isset($this->params[$name])) { continue;
$system->write(Json::encode($this->params[$name])); }
} $system = Snowflake::createObject($process, [Snowflake::app(), $name, true]);
$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
* @return Server
*/ /**
public function setDaemon($daemon): static * @param $daemon
{ * @return Server
if (!in_array($daemon, [0, 1])) { */
return $this; public function setDaemon($daemon): static
} {
$this->daemon = $daemon; if (!in_array($daemon, [0, 1])) {
return $this; return $this;
} }
$this->daemon = $daemon;
return $this;
/** }
* @return Packet|Websocket|Receive|Http|null
*/
public function getServer(): Packet|Websocket|Receive|Http|null /**
{ * @return Packet|Websocket|Receive|Http|null
return $this->baseServer; */
} public function getServer(): Packet|Websocket|Receive|Http|null
{
return $this->baseServer;
/** }
* @param $config
* @return mixed
* @throws Exception /**
*/ * @param $config
private function create($config): mixed * @return mixed
{ * @throws Exception
$settings = Config::get('settings', false, []); */
if (!isset($this->server[$config['type']])) { private function create($config): mixed
throw new Exception('Unknown server type(' . $config['type'] . ').'); {
} $settings = Config::get('settings', false, []);
$server = $this->dispatchCreate($config, $settings); if (!isset($this->server[$config['type']])) {
if (isset($config['events'])) { throw new Exception('Unknown server type(' . $config['type'] . ').');
$this->createEventListen($config); }
} $server = $this->dispatchCreate($config, $settings);
return $server; if (isset($config['events'])) {
} $this->createEventListen($config);
}
return $server;
/** }
* @param $config
* @throws Exception
*/ /**
protected function createEventListen($config) * @param $config
{ * @throws Exception
if (!is_array($config['events'])) { */
return; protected function createEventListen($config)
} {
$event = Snowflake::app()->getEvent(); if (!is_array($config['events'])) {
foreach ($config['events'] as $name => $_event) { return;
$event->on('listen ' . $config['port'] . ' ' . $name, $_event); }
} $event = Snowflake::app()->getEvent();
} foreach ($config['events'] as $name => $_event) {
$event->on('listen ' . $config['port'] . ' ' . $name, $_event);
/** }
* @param $config }
* @param $settings
* @return \Swoole\Server|Packet|Receive|Http|Websocket|null /**
* @throws NotFindClassException * @param $config
* @throws ReflectionException * @param $settings
* @throws Exception * @return \Swoole\Server|Packet|Receive|Http|Websocket|null
*/ * @throws NotFindClassException
private function dispatchCreate($config, $settings): \Swoole\Server|Packet|Receive|Http|Websocket|null * @throws ReflectionException
{ * @throws Exception
if (Snowflake::port_already($config['port'])) { */
return $this->error_stop($config['host'], $config['port']); private function dispatchCreate($config, $settings): \Swoole\Server|Packet|Receive|Http|Websocket|null
} {
if (!($this->baseServer instanceof \Swoole\Server)) { if (Snowflake::port_already($config['port'])) {
return $this->parseServer($config, $settings); return $this->error_stop($config['host'], $config['port']);
} }
return $this->addListener($config); if (!($this->baseServer instanceof \Swoole\Server)) {
} return $this->parseServer($config, $settings);
}
return $this->addListener($config);
/** }
* @param $config
* @return Http|Packet|Receive|Websocket|null
* @throws NotFindClassException /**
* @throws ReflectionException * @param $config
* @throws Exception * @return Http|Packet|Receive|Websocket|null
*/ * @throws NotFindClassException
private function addListener($config): Packet|Websocket|Receive|Http|null * @throws ReflectionException
{ * @throws Exception
$newListener = $this->baseServer->addlistener($config['host'], $config['port'], $config['mode']); */
if (!$newListener) { private function addListener($config): Packet|Websocket|Receive|Http|null
exit($this->addError(sprintf('Listen %s::%d fail.', $config['host'], $config['port']))); {
} $newListener = $this->baseServer->addlistener($config['host'], $config['port'], $config['mode']);
if (!$newListener) {
if (isset($config['settings']) && is_array($config['settings'])) { exit($this->addError(sprintf('Listen %s::%d fail.', $config['host'], $config['port'])));
$newListener->set($config['settings']); }
}
$this->onListenerBind($config, $this->baseServer); if (isset($config['settings']) && is_array($config['settings'])) {
$newListener->set($config['settings']);
return $this->baseServer; }
} $this->onListenerBind($config, $this->baseServer);
return $this->baseServer;
/** }
* @return Http|Packet|Receive|Websocket|null
* @throws ComponentException
* @throws ConfigException /**
* @throws NotFindClassException * @return Http|Packet|Receive|Websocket|null
* @throws ReflectionException * @throws ComponentException
*/ * @throws ConfigException
private function startRpcService(): Packet|Websocket|Receive|Http|null * @throws NotFindClassException
{ * @throws ReflectionException
$rpcService = Config::get('rpc.enable', false, []); */
if ($rpcService === true) { private function startRpcService(): Packet|Websocket|Receive|Http|null
/** @var Service $service */ {
$service = Snowflake::app()->get('rpc-service'); $rpcService = Config::get('rpc.enable', false, []);
$service->instance($this->baseServer); if ($rpcService === true) {
} /** @var Service $service */
return $this->baseServer; $service = Snowflake::app()->get('rpc-service');
} $service->instance($this->baseServer);
}
return $this->baseServer;
/** }
* @param $config
* @param $settings
* @return Packet|Websocket|Receive|Http|null /**
* @throws Exception * @param $config
*/ * @param $settings
private function parseServer($config, $settings): Packet|Websocket|Receive|Http|null * @return Packet|Websocket|Receive|Http|null
{ * @throws Exception
$class = $this->dispatch($config['type']); */
if (isset($config['settings']) && !empty($config['settings'])) { private function parseServer($config, $settings): Packet|Websocket|Receive|Http|null
$settings = array_merge($settings, $config['settings']); {
} $class = $this->dispatch($config['type']);
$this->baseServer = new $class($config['host'], $config['port'], SWOOLE_PROCESS, $config['mode']); if (isset($config['settings']) && !empty($config['settings'])) {
$settings['daemonize'] = $this->daemon; $settings = array_merge($settings, $config['settings']);
if (!isset($settings['pid_file'])) { }
$settings['pid_file'] = PID_PATH; $this->baseServer = new $class($config['host'], $config['port'], SWOOLE_PROCESS, $config['mode']);
} $settings['daemonize'] = $this->daemon;
$this->debug(Snowflake::listen($config)); if (!isset($settings['pid_file'])) {
$settings['pid_file'] = PID_PATH;
$this->onLoadWebsocketHandler(); }
if ($this->baseServer instanceof Http) { $this->debug(Snowflake::listen($config));
$this->onLoadHttpHandler();
} $this->onLoadWebsocketHandler();
if ($this->baseServer instanceof Http) {
$this->baseServer->set($settings); $this->onLoadHttpHandler();
}
$this->onProcessListener();
$this->baseServer->set($settings);
return $this->baseServer;
} $this->onProcessListener();
return $this->baseServer;
/** }
* @param $config
* @param $newListener
* @return Packet|Websocket|Receive|Http|null /**
* @throws NotFindClassException * @param $config
* @throws ReflectionException * @param $newListener
* @throws Exception * @return Packet|Websocket|Receive|Http|null
*/ * @throws NotFindClassException
private function onListenerBind($config, $newListener): Packet|Websocket|Receive|Http|null * @throws ReflectionException
{ * @throws Exception
if (!in_array($config['type'], [self::HTTP, self::TCP, self::PACKAGE])) { */
throw new Exception('Unknown server type(' . $config['type'] . ').'); private function onListenerBind($config, $newListener): Packet|Websocket|Receive|Http|null
} {
if (in_array($config['type'], $this->listenTypes)) { if (!in_array($config['type'], [self::HTTP, self::TCP, self::PACKAGE])) {
return $this->baseServer; throw new Exception('Unknown server type(' . $config['type'] . ').');
} }
if ($config['type'] == self::HTTP) { if (in_array($config['type'], $this->listenTypes)) {
$this->onBind($newListener, 'request', [Snowflake::createObject(OnRequest::class), 'onHandler']); return $this->baseServer;
} else { }
$this->noHttp($newListener, $config); if ($config['type'] == self::HTTP) {
} $this->onBind($newListener, 'request', [Snowflake::createObject(OnRequest::class), 'onHandler']);
$this->debug(sprintf('Check listen %s::%d -> ok', $config['host'], $config['port'])); } else {
$this->listenTypes[] = $config['type']; $this->noHttp($newListener, $config);
return $this->baseServer; }
} $this->debug(sprintf('Check listen %s::%d -> ok', $config['host'], $config['port']));
$this->listenTypes[] = $config['type'];
return $this->baseServer;
/** }
* @param $newListener
* @param $config
* @throws NotFindClassException /**
* @throws ReflectionException * @param $newListener
* @throws Exception * @param $config
*/ * @throws NotFindClassException
private function noHttp($newListener, $config) * @throws ReflectionException
{ * @throws Exception
$this->onBind($newListener, 'connect', [Snowflake::createObject(OnConnect::class), 'onHandler']); */
$this->onBind($newListener, 'close', [Snowflake::createObject(OnClose::class), 'onHandler']); private function noHttp($newListener, $config)
if ($config['type'] == self::TCP) { {
$this->onBind($newListener, 'receive', [$class = new OnReceive(), 'onHandler']); $this->onBind($newListener, 'connect', [Snowflake::createObject(OnConnect::class), 'onHandler']);
} else { $this->onBind($newListener, 'close', [Snowflake::createObject(OnClose::class), 'onHandler']);
$this->onBind($newListener, 'packet', [$class = new OnPacket(), 'onHandler']); if ($config['type'] == self::TCP) {
} $this->onBind($newListener, 'receive', [$class = new OnReceive(), 'onHandler']);
$class->host = $config['host']; } else {
$class->port = $config['port']; $this->onBind($newListener, 'packet', [$class = new OnPacket(), 'onHandler']);
} }
$class->host = $config['host'];
$class->port = $config['port'];
/** }
* @param $server
* @param $name
* @param $callback /**
* @throws Exception * @param $server
*/ * @param $name
private function onBind($server, $name, $callback) * @param $callback
{ * @throws Exception
if (in_array($name, $this->listening)) { */
return; private function onBind($server, $name, $callback)
} {
array_push($this->listening, $name); if (in_array($name, $this->listening)) {
if ($name === 'request') { return;
$this->onLoadHttpHandler(); }
} array_push($this->listening, $name);
$server->on($name, $callback); if ($name === 'request') {
} $this->onLoadHttpHandler();
}
$server->on($name, $callback);
/** }
* Load router handler
* @throws Exception
*/ /**
public function onLoadHttpHandler() * Load router handler
{ * @throws Exception
$event = Snowflake::app()->getEvent(); */
$event->on(Event::SERVER_WORKER_START, function () { public function onLoadHttpHandler()
router()->loadRouterSetting(); {
$event = Snowflake::app()->getEvent();
$annotation = Snowflake::app()->getAttributes(); $event->on(Event::SERVER_WORKER_START, function () {
$annotation->instanceDirectoryFiles(CONTROLLER_PATH); router()->loadRouterSetting();
$annotation->instanceDirectoryFiles(RPC_SERVICE_PATH);
$annotation->instanceDirectoryFiles(RPC_CLIENT_PATH); $annotation = Snowflake::app()->getAttributes();
}); $annotation->instanceDirectoryFiles(CONTROLLER_PATH);
} $annotation->instanceDirectoryFiles(RPC_SERVICE_PATH);
$annotation->instanceDirectoryFiles(RPC_CLIENT_PATH);
});
/** }
* @throws Exception
*/
public function onLoadWebsocketHandler() /**
{ * @throws Exception
$event = Snowflake::app()->getEvent(); */
$event->on(Event::SERVER_WORKER_START, function () { public function onLoadWebsocketHandler()
$annotation = Snowflake::app()->getAttributes(); {
$annotation->instanceDirectoryFiles(SOCKET_PATH); $event = Snowflake::app()->getEvent();
}); $event->on(Event::SERVER_WORKER_START, function () {
} $annotation = Snowflake::app()->getAttributes();
$annotation->instanceDirectoryFiles(SOCKET_PATH);
});
/** }
* @param $type
* @return string
*/ /**
private function dispatch($type): string * @param $type
{ * @return string
return match ($type) { */
self::HTTP => Http::class, private function dispatch($type): string
self::WEBSOCKET => Websocket::class, {
self::PACKAGE => Packet::class, return match ($type) {
default => Receive::class self::HTTP => Http::class,
}; self::WEBSOCKET => Websocket::class,
} self::PACKAGE => Packet::class,
default => Receive::class
/** };
* @param $servers }
* @return array
*/ /**
private function sortServers($servers): array * @param $servers
{ * @return array
$array = []; */
foreach ($servers as $server) { private function sortServers($servers): array
switch ($server['type']) { {
case self::WEBSOCKET: $array = [];
array_unshift($array, $server); foreach ($servers as $server) {
break; switch ($server['type']) {
case self::HTTP: case self::WEBSOCKET:
case self::PACKAGE | self::TCP: array_unshift($array, $server);
$array[] = $server; break;
break; case self::HTTP:
default: case self::PACKAGE | self::TCP:
$array[] = $server; $array[] = $server;
} break;
} default:
return $array; $array[] = $server;
} }
}
return $array;
}
} }