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

改名

Browse Source
This commit is contained in:
Mr·x
2021-08-01 11:18:09 +08:00
parent eb08fbaa0b
commit a0f1d7d933
1 changed files with 379 additions and 378 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Server/ServerManager.php
+379 -378
View File
@@ -25,437 +25,438 @@ use Swoole\WebSocket\Server as WServer;
class ServerManager extends Abstracts\Server class ServerManager extends Abstracts\Server
{ {
public string $host = ''; public string $host = '';
public int $port = 0; public int $port = 0;
/** @var Server\Port[] */ /** @var Server\Port[] */
public array $ports = []; public array $ports = [];
public int $mode = SWOOLE_TCP; public int $mode = SWOOLE_TCP;
private Server|WServer|HServer|null $server = null; private Server|WServer|HServer|null $server = null;
private static ?ServerManager $BASEServerListener = null; private static ?ServerManager $BASEServerListener = null;
/** /**
* @return static * @return static
*/ */
public static function getContext(): static public static function getContext(): static
{ {
if (!(static::$BASEServerListener)) { if (!(static::$BASEServerListener)) {
static::$BASEServerListener = new ServerManager(); static::$BASEServerListener = new ServerManager();
} }
return static::$BASEServerListener; return static::$BASEServerListener;
} }
/** /**
* @return Server|WServer|HServer|null * @return Server|WServer|HServer|null
*/ */
public function getServer(): Server|WServer|HServer|null public function getServer(): Server|WServer|HServer|null
{ {
return $this->server; return $this->server;
} }
/** /**
* @param string $type * @param string $type
* @param string $host * @param string $host
* @param int $port * @param int $port
* @param int $mode * @param int $mode
* @param array $settings * @param array $settings
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
* @throws ConfigException * @throws ConfigException
*/ * @throws Exception
public function addListener(string $type, string $host, int $port, int $mode, array $settings = []) */
{ public function addListener(string $type, string $host, int $port, int $mode, array $settings = [])
if ($this->checkPort($port)) $this->stopServer($port); {
if (!$this->server) { if ($this->checkPort($port)) $this->stopServer($port);
$this->createBaseServer($type, $host, $port, $mode, $settings); if (!$this->server) {
} else { $this->createBaseServer($type, $host, $port, $mode, $settings);
if (!isset($settings['settings'])) { } else {
$settings['settings'] = []; if (!isset($settings['settings'])) {
} $settings['settings'] = [];
$this->addNewListener($type, $host, $port, $mode, $settings); }
} $this->addNewListener($type, $host, $port, $mode, $settings);
} }
}
/** /**
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
*/ */
public function initBaseServer($configs): void public function initBaseServer($configs): void
{ {
$context = ServerManager::getContext(); $context = ServerManager::getContext();
foreach ($this->sortService($configs['ports']) as $config) { foreach ($this->sortService($configs['ports']) as $config) {
$this->startListenerHandler($context, $config); $this->startListenerHandler($context, $config);
} }
$this->addServerEventCallback($this->getSystemEvents($configs)); $this->addServerEventCallback($this->getSystemEvents($configs));
} }
/** /**
* @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);
} }
/** /**
* @param string|CustomProcess $customProcess * @param string|CustomProcess $customProcess
* @param null $redirect_stdin_and_stdout * @param null $redirect_stdin_and_stdout
* @param int|null $pipe_type * @param int|null $pipe_type
* @param bool $enable_coroutine * @param bool $enable_coroutine
* @throws Exception * @throws Exception
*/ */
public function addProcess(string|CustomProcess $customProcess, $redirect_stdin_and_stdout = null, ?int $pipe_type = SOCK_DGRAM, bool $enable_coroutine = true) public function addProcess(string|CustomProcess $customProcess, $redirect_stdin_and_stdout = null, ?int $pipe_type = SOCK_DGRAM, bool $enable_coroutine = true)
{ {
if (is_string($customProcess)) { if (is_string($customProcess)) {
$implements = class_implements($customProcess); $implements = class_implements($customProcess);
if (!in_array(CustomProcess::class, $implements)) { if (!in_array(CustomProcess::class, $implements)) {
trigger_error('custom process must implement ' . CustomProcess::class); trigger_error('custom process must implement ' . CustomProcess::class);
} }
$customProcess = new $customProcess($this->server); $customProcess = new $customProcess($this->server);
} }
/** @var Process $process */ /** @var Process $process */
$this->server->addProcess(new Process(function (Process $soloProcess) use ($customProcess) { $this->server->addProcess(new Process(function (Process $soloProcess) use ($customProcess) {
$system = sprintf('%s.process[%d]', Config::get('id', 'system-service'), $soloProcess->pid); $system = sprintf('%s.process[%d]', Config::get('id', 'system-service'), $soloProcess->pid);
if (Snowflake::getPlatform()->isLinux()) { if (Snowflake::getPlatform()->isLinux()) {
$soloProcess->name($system . '.' . $customProcess->getProcessName($soloProcess) . ' start.'); $soloProcess->name($system . '.' . $customProcess->getProcessName($soloProcess) . ' start.');
} }
echo sprintf("\033[36m[" . date('Y-m-d H:i:s') . "]\033[0m Process %s start.", $customProcess->getProcessName($soloProcess)) . PHP_EOL; echo sprintf("\033[36m[" . date('Y-m-d H:i:s') . "]\033[0m Process %s start.", $customProcess->getProcessName($soloProcess)) . PHP_EOL;
$customProcess->onHandler($soloProcess); $customProcess->onHandler($soloProcess);
}, },
$redirect_stdin_and_stdout, $pipe_type, $enable_coroutine)); $redirect_stdin_and_stdout, $pipe_type, $enable_coroutine));
} }
/** /**
* @param array $ports * @param array $ports
* @return array * @return array
*/ */
private function sortService(array $ports): array private function sortService(array $ports): array
{ {
$array = []; $array = [];
foreach ($ports as $port) { foreach ($ports as $port) {
if ($port['type'] == Constant::SERVER_TYPE_WEBSOCKET) { if ($port['type'] == Constant::SERVER_TYPE_WEBSOCKET) {
array_unshift($array, $port); array_unshift($array, $port);
} else if ($port['type'] == Constant::SERVER_TYPE_HTTP) { } else if ($port['type'] == Constant::SERVER_TYPE_HTTP) {
if (!empty($array) && $array[0]['type'] == Constant::SERVER_TYPE_WEBSOCKET) { if (!empty($array) && $array[0]['type'] == Constant::SERVER_TYPE_WEBSOCKET) {
$array[] = $port; $array[] = $port;
} else { } else {
array_unshift($array, $port); array_unshift($array, $port);
} }
} else { } else {
$array[] = $port; $array[] = $port;
} }
} }
return $array; return $array;
} }
/** /**
* @param array $configs * @param array $configs
* @return array * @return array
*/ */
private function getSystemEvents(array $configs): array private function getSystemEvents(array $configs): array
{ {
return array_intersect_key($configs['events'] ?? [], [ return array_intersect_key($configs['events'] ?? [], [
Constant::PIPE_MESSAGE => '', Constant::PIPE_MESSAGE => '',
Constant::SHUTDOWN => '', Constant::SHUTDOWN => '',
Constant::WORKER_START => '', Constant::WORKER_START => '',
Constant::WORKER_ERROR => '', Constant::WORKER_ERROR => '',
Constant::WORKER_EXIT => '', Constant::WORKER_EXIT => '',
Constant::WORKER_STOP => '', Constant::WORKER_STOP => '',
Constant::MANAGER_START => '', Constant::MANAGER_START => '',
Constant::MANAGER_STOP => '', Constant::MANAGER_STOP => '',
Constant::BEFORE_RELOAD => '', Constant::BEFORE_RELOAD => '',
Constant::AFTER_RELOAD => '', Constant::AFTER_RELOAD => '',
Constant::DISCONNECT => '', Constant::DISCONNECT => '',
Constant::START => '', Constant::START => '',
]); ]);
} }
/** /**
* @param ServerManager $context * @param ServerManager $context
* @param array $config * @param array $config
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
* @throws Exception * @throws Exception
*/ */
private function startListenerHandler(ServerManager $context, array $config) private function startListenerHandler(ServerManager $context, array $config)
{ {
if ($this->server) { if ($this->server) {
$context->addListener($config['type'], $config['host'], $config['port'], $config['mode'], $config); $context->addListener($config['type'], $config['host'], $config['port'], $config['mode'], $config);
} else { } else {
$config['settings'] = $config['settings'] ?? []; $config['settings'] = $config['settings'] ?? [];
if (!isset($config['settings']['log_file'])) { if (!isset($config['settings']['log_file'])) {
$config['settings']['log_file'] = storage('system.log'); $config['settings']['log_file'] = storage('system.log');
} }
if (!isset($config['settings']['pid_file'])) { if (!isset($config['settings']['pid_file'])) {
$config['settings']['pid_file'] = storage('swoole.pid'); $config['settings']['pid_file'] = storage('swoole.pid');
} }
$config['events'] = $config['events'] ?? []; $config['events'] = $config['events'] ?? [];
$context->addListener($config['type'], $config['host'], $config['port'], $config['mode'], $config); $context->addListener($config['type'], $config['host'], $config['port'], $config['mode'], $config);
} }
} }
/** /**
* @param string $type * @param string $type
* @param string $host * @param string $host
* @param int $port * @param int $port
* @param int $mode * @param int $mode
* @param array $settings * @param array $settings
* @throws ReflectionException * @throws ReflectionException
* @throws NotFindClassException * @throws NotFindClassException
*/ */
private function addNewListener(string $type, string $host, int $port, int $mode, array $settings = []) private function addNewListener(string $type, string $host, int $port, int $mode, array $settings = [])
{ {
echo sprintf("\033[36m[" . date('Y-m-d H:i:s') . "]\033[0m $type service %s::%d start.", $host, $port) . PHP_EOL; echo sprintf("\033[36m[" . date('Y-m-d H:i:s') . "]\033[0m $type service %s::%d start.", $host, $port) . PHP_EOL;
switch ($type) { switch ($type) {
case Constant::SERVER_TYPE_TCP: case Constant::SERVER_TYPE_TCP:
$this->ports[$port] = TCPServerListener::instance($this->server, $host, $port, $mode, $settings); $this->ports[$port] = TCPServerListener::instance($this->server, $host, $port, $mode, $settings);
break; break;
case Constant::SERVER_TYPE_UDP: case Constant::SERVER_TYPE_UDP:
$this->ports[$port] = UDPServerListener::instance($this->server, $host, $port, $mode, $settings); $this->ports[$port] = UDPServerListener::instance($this->server, $host, $port, $mode, $settings);
break; break;
case Constant::SERVER_TYPE_HTTP: case Constant::SERVER_TYPE_HTTP:
$this->ports[$port] = HTTPServerListener::instance($this->server, $host, $port, $mode, $settings); $this->ports[$port] = HTTPServerListener::instance($this->server, $host, $port, $mode, $settings);
break; break;
case Constant::SERVER_TYPE_WEBSOCKET: case Constant::SERVER_TYPE_WEBSOCKET:
$this->ports[$port] = WebSocketServerListener::instance($this->server, $host, $port, $mode, $settings); $this->ports[$port] = WebSocketServerListener::instance($this->server, $host, $port, $mode, $settings);
break; break;
} }
} }
/** /**
* @param int $port * @param int $port
* @param string $event * @param string $event
* @return Closure|array|null * @return Closure|array|null
*/ */
public function getPortCallback(int $port, string $event): Closure|array|null public function getPortCallback(int $port, string $event): Closure|array|null
{ {
/** @var Server\Port $_port */ /** @var Server\Port $_port */
$_port = $this->ports[$port] ?? null; $_port = $this->ports[$port] ?? null;
if (is_null($_port)) { if (is_null($_port)) {
return null; return null;
} }
return $_port->getCallback($event); return $_port->getCallback($event);
} }
/** /**
* @param string $type * @param string $type
* @param string $host * @param string $host
* @param int $port * @param int $port
* @param int $mode * @param int $mode
* @param array $settings * @param array $settings
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
* @throws ConfigException * @throws ConfigException
*/ */
private function createBaseServer(string $type, string $host, int $port, int $mode, array $settings = []) private function createBaseServer(string $type, string $host, int $port, int $mode, array $settings = [])
{ {
$match = match ($type) { $match = match ($type) {
Constant::SERVER_TYPE_BASE, Constant::SERVER_TYPE_TCP, Constant::SERVER_TYPE_BASE, Constant::SERVER_TYPE_TCP,
Constant::SERVER_TYPE_UDP => Server::class, Constant::SERVER_TYPE_UDP => Server::class,
Constant::SERVER_TYPE_HTTP => HServer::class, Constant::SERVER_TYPE_HTTP => HServer::class,
Constant::SERVER_TYPE_WEBSOCKET => WServer::class Constant::SERVER_TYPE_WEBSOCKET => WServer::class
}; };
$this->server = new $match($host, $port, SWOOLE_PROCESS, $mode); $this->server = new $match($host, $port, SWOOLE_PROCESS, $mode);
$this->server->set(array_merge(Config::get('server.settings', []), $settings['settings'])); $this->server->set(array_merge(Config::get('server.settings', []), $settings['settings']));
echo sprintf("\033[36m[" . date('Y-m-d H:i:s') . "]\033[0m $type service %s::%d start.", $host, $port) . PHP_EOL; echo sprintf("\033[36m[" . date('Y-m-d H:i:s') . "]\033[0m $type service %s::%d start.", $host, $port) . PHP_EOL;
$this->addDefaultListener($type, $settings); $this->addDefaultListener($type, $settings);
} }
/** /**
* @param int $port * @param int $port
*/ */
public function stopServer(int $port) public function stopServer(int $port)
{ {
if (!($pid = $this->portIsAready($port))) { if (!($pid = $this->portIsAready($port))) {
return; return;
} }
exec('kill -15 ' . $pid, $execResult); exec('kill -15 ' . $pid, $execResult);
while ($this->portIsAready($port)) { while ($this->portIsAready($port)) {
usleep(100); usleep(100);
} }
} }
/** /**
* @param $port * @param $port
* @return bool * @return bool|string
*/ */
private function portIsAready($port): bool|string private function portIsAready($port): bool|string
{ {
exec('netstat -lnp | grep ' . $port . ' | grep "LISTEN" | awk \'{print $7}\'', $output); exec('netstat -lnp | grep ' . $port . ' | grep "LISTEN" | awk \'{print $7}\'', $output);
if (empty($output)) { if (empty($output)) {
return false; return false;
} }
return explode('/', $output[0])[0]; return explode('/', $output[0])[0];
} }
/** /**
* @param string $type * @param string $type
* @param array $settings * @param array $settings
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
*/ */
private function addDefaultListener(string $type, array $settings): void private function addDefaultListener(string $type, array $settings): void
{ {
$this->addServerEventCallback($settings['events']); $this->addServerEventCallback($settings['events']);
if (($this->server->setting['task_worker_num'] ?? 0) > 0) { if (($this->server->setting['task_worker_num'] ?? 0) > 0) {
$this->addTaskListener($settings['events']); $this->addTaskListener($settings['events']);
} }
if ($type === Constant::SERVER_TYPE_WEBSOCKET) { if ($type === Constant::SERVER_TYPE_WEBSOCKET) {
$reflect = $this->getNewInstance(WebSocketServerListener::class); $reflect = $this->getNewInstance(WebSocketServerListener::class);
$this->server->on('handshake', [$reflect, 'onHandshake']); $this->server->on('handshake', [$reflect, 'onHandshake']);
$this->server->on('message', [$reflect, 'onMessage']); $this->server->on('message', [$reflect, 'onMessage']);
$this->server->on('close', [$reflect, 'onClose']); $this->server->on('close', [$reflect, 'onClose']);
$reflect->setEvents(Constant::HANDSHAKE, $settings['events'][Constant::HANDSHAKE] ?? null); $reflect->setEvents(Constant::HANDSHAKE, $settings['events'][Constant::HANDSHAKE] ?? null);
$reflect->setEvents(Constant::MESSAGE, $settings['events'][Constant::MESSAGE] ?? null); $reflect->setEvents(Constant::MESSAGE, $settings['events'][Constant::MESSAGE] ?? null);
$reflect->setEvents(Constant::CONNECT, $settings['events'][Constant::CONNECT] ?? null); $reflect->setEvents(Constant::CONNECT, $settings['events'][Constant::CONNECT] ?? null);
$reflect->setEvents(Constant::CLOSE, $settings['events'][Constant::CLOSE] ?? null); $reflect->setEvents(Constant::CLOSE, $settings['events'][Constant::CLOSE] ?? null);
} else if ($type === Constant::SERVER_TYPE_UDP) { } else if ($type === Constant::SERVER_TYPE_UDP) {
$reflect = $this->getNewInstance(UDPServerListener::class); $reflect = $this->getNewInstance(UDPServerListener::class);
$this->server->on('packet', [$reflect, 'onPacket']); $this->server->on('packet', [$reflect, 'onPacket']);
$reflect->setEvents(Constant::PACKET, $settings['events'][Constant::PACKET] ?? null); $reflect->setEvents(Constant::PACKET, $settings['events'][Constant::PACKET] ?? null);
} else if ($type === Constant::SERVER_TYPE_HTTP) { } else if ($type === Constant::SERVER_TYPE_HTTP) {
$reflect = $this->getNewInstance(HTTPServerListener::class); $reflect = $this->getNewInstance(HTTPServerListener::class);
$this->server->on('request', [$reflect, 'onRequest']); $this->server->on('request', [$reflect, 'onRequest']);
$this->addCloseOrDisconnect($reflect, $settings); $this->addCloseOrDisconnect($reflect, $settings);
} else { } else {
$reflect = $this->getNewInstance(TCPServerListener::class); $reflect = $this->getNewInstance(TCPServerListener::class);
$this->server->on('connect', [$reflect, 'onConnect']); $this->server->on('connect', [$reflect, 'onConnect']);
$this->server->on('receive', [$reflect, 'onReceive']); $this->server->on('receive', [$reflect, 'onReceive']);
$this->addCloseOrDisconnect($reflect, $settings); $this->addCloseOrDisconnect($reflect, $settings);
} }
} }
/** /**
* *
*/ */
public function start() public function start()
{ {
$this->server->start(); $this->server->start();
} }
/** /**
* @param string $class * @param string $class
* @return object * @return object
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
*/ */
private function getNewInstance(string $class): object private function getNewInstance(string $class): object
{ {
return Snowflake::getDi()->getReflect($class)?->newInstance(); return Snowflake::getDi()->getReflect($class)?->newInstance();
} }
/** /**
* @param $reflect * @param $reflect
* @param $settings * @param $settings
*/ */
private function addCloseOrDisconnect($reflect, $settings): void private function addCloseOrDisconnect($reflect, $settings): void
{ {
if (swoole_version() >= '4.7.0') { if (swoole_version() >= '4.7.0') {
$this->server->on('disconnect', [$reflect, 'onDisconnect']); $this->server->on('disconnect', [$reflect, 'onDisconnect']);
$reflect->setEvents(Constant::DISCONNECT, $settings['events'][Constant::DISCONNECT] ?? null); $reflect->setEvents(Constant::DISCONNECT, $settings['events'][Constant::DISCONNECT] ?? null);
} else { } else {
$this->server->on('close', [$reflect, 'onClose']); $this->server->on('close', [$reflect, 'onClose']);
$reflect->setEvents(Constant::CLOSE, $settings['events'][Constant::CLOSE] ?? null); $reflect->setEvents(Constant::CLOSE, $settings['events'][Constant::CLOSE] ?? null);
} }
} }
/** /**
* @param array $events * @param array $events
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
*/ */
private function addServerEventCallback(array $events): void private function addServerEventCallback(array $events): void
{ {
if (count($events) < 1) { if (count($events) < 1) {
return; return;
} }
foreach ($events as $event_type => $callback) { foreach ($events as $event_type => $callback) {
if ($this->server->getCallback($event_type) !== null) { if ($this->server->getCallback($event_type) !== null) {
continue; continue;
} }
if (is_array($callback) && !is_object($callback[0])) { if (is_array($callback) && !is_object($callback[0])) {
$callback[0] = Snowflake::getDi()->get($callback[0]); $callback[0] = Snowflake::getDi()->get($callback[0]);
} }
$this->server->on($event_type, $callback); $this->server->on($event_type, $callback);
} }
} }
/** /**
* @param TaskExecute|string $handler * @param TaskExecute|string $handler
* @param array $params * @param array $params
* @param int $workerId * @param int $workerId
* @throws Exception * @throws Exception
*/ */
public function task(TaskExecute|string $handler, array $params = [], int $workerId = 0) public function task(TaskExecute|string $handler, array $params = [], int $workerId = 0)
{ {
if ($workerId === null || $workerId <= $this->server->setting['worker_num']) { if ($workerId === null || $workerId <= $this->server->setting['worker_num']) {
$workerId = random_int($this->server->setting['worker_num'] + 1, $workerId = random_int($this->server->setting['worker_num'] + 1,
$this->server->setting['worker_num'] + 1 + $this->server->setting['task_worker_num']); $this->server->setting['worker_num'] + 1 + $this->server->setting['task_worker_num']);
} }
if (is_string($handler)) { if (is_string($handler)) {
$implements = Snowflake::getDi()->getReflect($handler); $implements = Snowflake::getDi()->getReflect($handler);
if (!in_array(TaskExecute::class, $implements->getInterfaceNames())) { if (!in_array(TaskExecute::class, $implements->getInterfaceNames())) {
throw new Exception('Task must instance ' . TaskExecute::class); throw new Exception('Task must instance ' . TaskExecute::class);
} }
$handler = $implements->newInstanceArgs($params); $handler = $implements->newInstanceArgs($params);
} }
$this->server->task(serialize($handler), $workerId); $this->server->task(serialize($handler), $workerId);
} }
/** /**
* @param array $events * @param array $events
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
*/ */
private function addTaskListener(array $events = []): void private function addTaskListener(array $events = []): void
{ {
$task_use_object = $this->server->setting['task_object'] ?? $this->server->setting['task_use_object'] ?? false; $task_use_object = $this->server->setting['task_object'] ?? $this->server->setting['task_use_object'] ?? false;
$reflect = Snowflake::getDi()->getReflect(ServerTask::class)?->newInstance(); $reflect = Snowflake::getDi()->getReflect(ServerTask::class)?->newInstance();
if ($task_use_object || $this->server->setting['task_enable_coroutine']) { if ($task_use_object || $this->server->setting['task_enable_coroutine']) {
$this->server->on('task', $events[Constant::TASK] ?? [$reflect, 'onCoroutineTask']); $this->server->on('task', $events[Constant::TASK] ?? [$reflect, 'onCoroutineTask']);
} else { } else {
$this->server->on('task', $events[Constant::TASK] ?? [$reflect, 'onTask']); $this->server->on('task', $events[Constant::TASK] ?? [$reflect, 'onTask']);
} }
$this->server->on('finish', $events[Constant::FINISH] ?? [$reflect, 'onFinish']); $this->server->on('finish', $events[Constant::FINISH] ?? [$reflect, 'onFinish']);
} }
} }