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-07-19 19:04:13 +08:00
parent 5aa94c1a25
commit b369e5c492
21 changed files with 952 additions and 616 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Annotation/Port.php
+2 -1
View File
@@ -27,7 +27,8 @@ use Snowflake\Snowflake;
/**
* @param array $handler
* @param mixed $class
* @param mixed|null $method
* @return mixed
* @throws Exception
*/
HttpServer/Route/CoreMiddleware.php
+5
View File
@@ -21,6 +21,11 @@ use Snowflake\Snowflake;
class CoreMiddleware implements Middleware
{
/** @var int */
public int $zOrder = 0;
/**
* @param Request $request
* @param Closure $next
HttpServer/Route/Router.php
+1 -1
View File
@@ -546,7 +546,7 @@ class Router extends HttpService implements RouterInterface
*/
public function find_path(Request $request): ?Node
{
return $this->Branch_search($request);
// return $this->Branch_search($request);
$method = $request->getMethod();
$uri = $request->headers->get('request_uri', '/');
Rpc/Actuator.php
+3 -2
View File
@@ -4,6 +4,7 @@
namespace Rpc;
use Exception;
use HttpServer\Route\Router;
use Snowflake\Snowflake;
@@ -22,7 +23,7 @@ class Actuator
/**
* Actuator constructor.
* @param int $port
* @throws \Exception
* @throws Exception
*/
public function __construct(public int $port)
{
@@ -33,7 +34,7 @@ class Actuator
/**
* @param string $path
* @param string|callable $callback
* @throws \Exception
* @throws Exception
*/
public function addListener(string $path, string|callable $callback): void
{
Rpc/Service.php
+6 -4
View File
@@ -77,10 +77,12 @@ class Service extends Component
}
/**
* @param $mode
* @throws Exception
*/
/**
* @param $rpcServer
* @param $config
* @param $mode
* @throws Exception
*/
private function addCallback($rpcServer, $config, $mode)
{
$tcp = [SWOOLE_SOCK_TCP, SWOOLE_TCP, SWOOLE_TCP6, SWOOLE_SOCK_TCP6];
Server/Abstracts/Server.php
+82
View File
@@ -0,0 +1,82 @@
<?php
namespace Server\Abstracts;
use Closure;
use Exception;
use Snowflake\Snowflake;
/**
* Class Server
* @package Server\Abstracts
*/
abstract class Server
{
protected array $_events = [];
/**
* @param string $name
* @param array|null $events
* @throws Exception
*/
public function setEvents(string $name, ?array $events): void
{
if (!is_callable($events)) {
return;
}
if (is_array($events) && is_string($events[0])) {
$reflect = Snowflake::getDi()->getReflect($events[0]);
if (!$reflect) {
throw new Exception('Checks the class is c\'not instantiable.');
}
$events[0] = $reflect->newInstance();
}
$this->_events[$name] = $events;
}
/**
* @return array
*/
public function getEvents(): array
{
return $this->_events;
}
/**
* @param string $name
* @return mixed
*/
public function getEvent(string $name): mixed
{
return $this->_events[$name] ?? null;
}
/**
* @param $name
* @param Closure|null $closure
* @param array $params
* @return mixed
*/
public function runEvent($name, ?Closure $closure, array $params): void
{
$event = $this->getEvent($name);
if (empty($event)) {
if (!is_callable($closure)) {
return;
}
call_user_func($closure, ...$params);
} else {
call_user_func($event, ...$params);
}
}
}
Server/BASEServerListener.php
-289
View File
@@ -1,289 +0,0 @@
<?php
namespace Server;
use Server\SInterface\CustomProcess;
use Snowflake\Snowflake;
use Swoole\Http\Server as HServer;
use Swoole\Process;
use Swoole\Server;
use Swoole\WebSocket\Server as WServer;
use Server\Task\ServerTask;
/**
* Class BASEServerListener
* @package HttpServer\Service
*/
class BASEServerListener
{
public string $host = '';
public int $port = 0;
public int $mode = SWOOLE_TCP;
private Server|WServer|HServer|null $server = null;
private static ?BASEServerListener $BASEServerListener = null;
/**
* @return static
*/
public static function getContext(): static
{
if (!(static::$BASEServerListener)) {
static::$BASEServerListener = new BASEServerListener();
}
return static::$BASEServerListener;
}
/**
* @return Server|WServer|HServer|null
*/
public function getServer(): Server|WServer|HServer|null
{
return $this->server;
}
/**
* @param string $type
* @param string $host
* @param int $port
* @param int $mode
* @param array $settings
*/
public function addListener(string $type, string $host, int $port, int $mode, array $settings = [])
{
if (!$this->server) {
$this->createBaseServer($type, $host, $port, $mode, $settings);
} else {
if (!isset($settings['settings'])) {
$settings['settings'] = [];
}
$this->addNewListener($type, $host, $port, $mode, $settings);
}
}
/**
* startRun
*/
public function start(): void
{
$context = BASEServerListener::getContext();
$configs = require_once 'server.php';
foreach ($this->sortService($configs['server']['ports']) as $config) {
$this->startListenerHandler($context, $config);
}
$this->addServerEventCallback($this->getSystemEvents($configs));
$context->server->start();
}
/**
* @param string|CustomProcess $customProcess
* @param null $redirect_stdin_and_stdout
* @param int|null $pipe_type
* @param bool $enable_coroutine
*/
public function addProcess(string|CustomProcess $customProcess, $redirect_stdin_and_stdout = null, ?int $pipe_type = SOCK_DGRAM, bool $enable_coroutine = true)
{
if (is_string($customProcess)) {
$implements = class_implements($customProcess);
if (!in_array(CustomProcess::class, $implements)) {
trigger_error('custom process must implement ' . CustomProcess::class);
}
$customProcess = new $customProcess($this->server);
}
/** @var Process $process */
$this->server->addProcess(new Process(function (Process $soloProcess) use ($customProcess) {
$soloProcess->name($customProcess->getProcessName($soloProcess));
$customProcess->onHandler($soloProcess);
},
$redirect_stdin_and_stdout, $pipe_type, $enable_coroutine));
}
/**
* @param array $ports
* @return array
*/
private function sortService(array $ports): array
{
$array = [];
foreach ($ports as $port) {
if ($port['type'] == Constant::SERVER_TYPE_WEBSOCKET) {
array_unshift($array, $port);
} else if ($port['type'] == Constant::SERVER_TYPE_HTTP) {
if (!empty($array) && $array[0]['type'] == Constant::SERVER_TYPE_WEBSOCKET) {
$array[] = $port;
} else {
array_unshift($array, $port);
}
} else {
$array[] = $port;
}
}
return $array;
}
/**
* @param array $configs
* @return array
*/
private function getSystemEvents(array $configs): array
{
return array_intersect_key($configs['server']['events'] ?? [], [
Constant::PIPE_MESSAGE => '',
Constant::SHUTDOWN => '',
Constant::WORKER_START => '',
Constant::WORKER_ERROR => '',
Constant::WORKER_EXIT => '',
Constant::WORKER_STOP => '',
Constant::MANAGER_START => '',
Constant::MANAGER_STOP => '',
Constant::BEFORE_RELOAD => '',
Constant::AFTER_RELOAD => '',
Constant::START => '',
]);
}
/**
* @param BASEServerListener $context
* @param array $config
*/
private function startListenerHandler(BASEServerListener $context, array $config)
{
if ($this->server) {
$context->addNewListener($config['type'], $config['host'], $config['port'], $config['mode'], $config);
} else {
$config['settings'] = array_merge($configs['settings'] ?? [], $config['settings'] ?? []);
$config['events'] = array_merge($configs['events'] ?? [], $config['events'] ?? []);
$context->createBaseServer($config['type'], $config['host'], $config['port'], $config['mode'], $config);
}
}
/**
* @param string $type
* @param string $host
* @param int $port
* @param int $mode
* @param array $settings
*/
private function addNewListener(string $type, string $host, int $port, int $mode, array $settings = [])
{
switch ($type) {
case Constant::SERVER_TYPE_TCP:
TCPServerListener::instance($this->server, $host, $port, $mode, $settings);
break;
case Constant::SERVER_TYPE_UDP:
UDPServerListener::instance($this->server, $host, $port, $mode, $settings);
break;
case Constant::SERVER_TYPE_HTTP:
HTTPServerListener::instance($this->server, $host, $port, $mode, $settings);
break;
case Constant::SERVER_TYPE_WEBSOCKET:
WebSocketServerListener::instance($this->server, $host, $port, $mode, $settings);
break;
}
}
/**
* @param string $type
* @param string $host
* @param int $port
* @param int $mode
* @param array $settings
*/
private function createBaseServer(string $type, string $host, int $port, int $mode, array $settings = [])
{
$match = match ($type) {
Constant::SERVER_TYPE_BASE,
Constant::SERVER_TYPE_TCP,
Constant::SERVER_TYPE_UDP => Server::class,
Constant::SERVER_TYPE_HTTP => HServer::class,
Constant::SERVER_TYPE_WEBSOCKET => WServer::class
};
$this->server = new $match($host, $port, SWOOLE_PROCESS, $mode);
$this->server->set($settings['settings']);
$this->addDefaultListener($type, $settings);
}
/**
* @param string $type
* @param array $settings
* @return void
*/
private function addDefaultListener(string $type, array $settings): void
{
if (($this->server->setting['task_worker_num'] ?? 0) > 0) $this->addTaskListener($settings['events']);
if ($type === Constant::SERVER_TYPE_WEBSOCKET) {
$reflect = Snowflake::getDi()->getReflect(WebSocketServerListener::class)?->newInstance();
$this->server->on('handshake', $settings['events'][Constant::HANDSHAKE] ?? [$reflect, 'onHandshake']);
$this->server->on('message', $settings['events'][Constant::MESSAGE] ?? [$reflect, 'onMessage']);
$this->server->on('close', $settings['events'][Constant::CLOSE] ?? [$reflect, 'onClose']);
} else if ($type === Constant::SERVER_TYPE_UDP) {
$reflect = Snowflake::getDi()->getReflect(UDPServerListener::class)?->newInstance();
$this->server->on('packet', $settings['events'][Constant::PACKET] ?? [$reflect, 'onPacket']);
} else if ($type === Constant::SERVER_TYPE_HTTP) {
$reflect = Snowflake::getDi()->getReflect(HTTPServerListener::class)?->newInstance();
$this->server->on('request', $settings['events'][Constant::REQUEST] ?? [$reflect, 'onRequest']);
} else {
$reflect = Snowflake::getDi()->getReflect(TCPServerListener::class)?->newInstance();
$this->server->on('receive', $settings['events'][Constant::RECEIVE] ?? [$reflect, 'onReceive']);
}
$this->addServerEventCallback($settings['events']);
}
/**
* @param array $events
*/
private function addServerEventCallback(array $events)
{
if (count($events) < 1) {
return;
}
foreach ($events as $event_type => $callback) {
if ($this->server->getCallback($event_type) !== null) {
continue;
}
$this->server->on($event_type, $callback);
}
}
/**
* @param array $events
*/
private function addTaskListener(array $events = []): void
{
$task_use_object = $this->server->setting['task_object'] ?? $this->server->setting['task_use_object'] ?? false;
$reflect = Snowflake::getDi()->getReflect(ServerTask::class)?->newInstance();
if ($task_use_object || $this->server->setting['task_enable_coroutine']) {
$this->server->on('task', $events[Constant::TASK] ?? [$reflect, 'onCoroutineTask']);
} else {
$this->server->on('task', $events[Constant::TASK] ?? [$reflect, 'onTask']);
}
$this->server->on('finish', $events[Constant::FINISH] ?? [$reflect, 'onFinish']);
}
}
$context = BASEServerListener::getContext();
$context->start();
Server/Constant.php
+5
View File
@@ -4,6 +4,10 @@
namespace Server;
/**
* Class Constant
* @package Server
*/
class Constant
{
@@ -14,6 +18,7 @@ class Constant
const WORKER_EXIT = 'WorkerExit';
const CONNECT = 'Connect';
const HANDSHAKE = 'handshake';
const DISCONNECT = 'disconnect';
const MESSAGE = 'message';
const RECEIVE = 'Receive';
const PACKET = 'Packet';
Server/HTTPServerListener.php
+72 -52
View File
@@ -3,73 +3,93 @@
namespace Server;
use HttpServer\Route\Router;
use ReflectionException;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;
use Swoole\Http\Request;
use Swoole\Http\Response;
use Swoole\Server;
class HTTPServerListener
/**
* Class HTTPServerListener
* @package Server
*/
class HTTPServerListener extends Abstracts\Server
{
protected static mixed $_http;
protected static mixed $_http;
use ListenerHelper;
use ListenerHelper;
private Router $router;
private Router $router;
/**
* UDPServerListener constructor.
* @param Server|\Swoole\WebSocket\Server|\Swoole\Http\Server $server
* @param string $host
* @param int $port
* @param int $mode
* @param array|null $settings
* @throws \ReflectionException
*/
public static function instance(mixed $server, string $host, int $port, int $mode, ?array $settings = [])
{
if (!in_array($mode, [SWOOLE_TCP, SWOOLE_TCP6])) {
trigger_error('Port mode ' . $host . '::' . $port . ' must is udp listener type.');
}
$reflect = Snowflake::getDi()->getReflect(static::class)?->newInstance();
static::$_http = $server->addlistener($host, $port, $mode);
static::$_http->set($settings['settings'] ?? []);
static::$_http->on('request', self::callback(Constant::REQUEST, $settings['events'], [$reflect, 'onRequest']));
static::$_http->on('connect', static::callback(Constant::CONNECT, $settings['events'], [$reflect, 'onConnect']));
static::$_http->on('close', static::callback(Constant::CLOSE, $settings['events'], [$reflect, 'onClose']));
}
/**
* UDPServerListener constructor.
* @param Server|\Swoole\WebSocket\Server|\Swoole\Http\Server $server
* @param string $host
* @param int $port
* @param int $mode
* @param array|null $settings
* @return Server\Port
* @throws NotFindClassException
* @throws ReflectionException
* @throws \Exception
*/
public static function instance(mixed $server, string $host, int $port, int $mode, ?array $settings = []): Server\Port
{
if (!in_array($mode, [SWOOLE_TCP, SWOOLE_TCP6])) {
trigger_error('Port mode ' . $host . '::' . $port . ' must is udp listener type.');
}
/** @var static $reflect */
$reflect = Snowflake::getDi()->getReflect(static::class)?->newInstance();
static::$_http = $server->addlistener($host, $port, $mode);
static::$_http->set($settings['settings'] ?? []);
static::$_http->on('request', [$reflect, 'onRequest']);
static::$_http->on('connect', [$reflect, 'onConnect']);
static::$_http->on('disconnect', [$reflect, 'onDisconnect']);
$reflect->setEvents(Constant::DISCONNECT, $settings['events'][Constant::DISCONNECT] ?? null);
$reflect->setEvents(Constant::CONNECT, $settings['events'][Constant::CONNECT] ?? null);
return static::$_http;
}
/**
* @param Server $server
* @param int $fd
*/
public function onConnect(Server $server, int $fd)
{
$server->confirm($fd);
}
/**
* @param Server $server
* @param int $fd
*/
public function onConnect(Server $server, int $fd)
{
$this->runEvent(Constant::CONNECT, null, [$server, $fd]);
}
/**
* @param Request $request
* @param Response $response
*/
public function onRequest(Request $request, Response $response)
{
if (!$response->isWritable()) {
return;
}
$response->status(200);
$response->end('');
}
/**
* @param Request $request
* @param Response $response
*/
public function onRequest(Request $request, Response $response)
{
$this->router->find_path(new \HttpServer\Http\Request());
if (!$response->isWritable()) {
return;
}
$response->status(200);
$response->end('');
}
/**
* @param Server $server
* @param int $fd
*/
public function onClose(Server $server, int $fd)
{
}
/**
* @param Server $server
* @param int $fd
*/
public function onDisconnect(Server $server, int $fd)
{
$this->runEvent(Constant::CONNECT, null, [$server, $fd]);
}
}
Server/ListenerHelper.php
+18 -23
View File
@@ -3,41 +3,36 @@
namespace Server;
use Closure;
use ReflectionException;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;
/**
* Trait ListenerHelper
* @package Server
*/
trait ListenerHelper
{
/**
* @param $server
* @param $newServer
*/
public static function onConnectAndClose($settings, $newServer)
{
$reflect = Snowflake::getDi()->getReflect(static::class)?->newInstance();
$newServer->on('connect', static::callback(Constant::CONNECT, $settings['events'], [$reflect, 'onConnect']));
$newServer->on('close', static::callback(Constant::CLOSE, $settings['events'], [$reflect, 'onClose']));
}
/**
* @param string $name
* @param array $events
* @param array|\Closure $default
* @return array|\Closure|mixed
* @throws \ReflectionException
* @throws \Snowflake\Exception\NotFindClassException
*/
protected static function callback(string $name, array $events, array|\Closure $default)
/**
* @param string $name
* @param array $events
* @param array|Closure $default
* @return mixed
* @throws NotFindClassException
* @throws ReflectionException
*/
protected static function callback(string $name, array $events, array|Closure $default): mixed
{
if (!is_array($events) || !isset($events[$name])) {
return $default;
}
$callback = $events[$name];
if ($callback instanceof \Closure) {
if ($callback instanceof Closure) {
return $callback;
}
$object = Snowflake::getDi()->getReflect($callback[0]);
Server/Manager/ServerBase.php
+36 -11
View File
@@ -2,37 +2,62 @@
namespace Server\Manager;
class ServerBase
use Server\Abstracts\Server;
use Server\Constant;
/**
* Class ServerBase
* @package Server\Manager
*/
class ServerBase extends Server
{
public function onStart()
/**
* @param \Swoole\Server $server
*/
public function onStart(\Swoole\Server $server)
{
var_dump(func_get_args());
$this->runEvent(Constant::START, null, [$server]);
}
public function onShutdown()
/**
* @param \Swoole\Server $server
*/
public function onShutdown(\Swoole\Server $server)
{
var_dump(func_get_args());
$this->runEvent(Constant::SHUTDOWN, null, [$server]);
}
public function onPipeMessage()
/**
* @param \Swoole\Server $server
* @param int $src_worker_id
* @param mixed $message
*/
public function onPipeMessage(\Swoole\Server $server, int $src_worker_id, mixed $message)
{
$this->runEvent(Constant::PIPE_MESSAGE, null, [$server, $src_worker_id, $message]);
}
public function onBeforeReload()
/**
* @param \Swoole\Server $server
*/
public function onBeforeReload(\Swoole\Server $server)
{
$this->runEvent(Constant::PIPE_MESSAGE, null, [$server]);
}
public function onAfterReload()
/**
* @param \Swoole\Server $server
*/
public function onAfterReload(\Swoole\Server $server)
{
$this->runEvent(Constant::PIPE_MESSAGE, null, [$server]);
}
}
Server/Manager/ServerManager.php
+20 -5
View File
@@ -2,18 +2,33 @@
namespace Server\Manager;
class ServerManager
use Server\Abstracts\Server;
use Server\Constant;
/**
* Class ServerManager
* @package Server\Manager
*/
class ServerManager extends Server
{
public function onManagerStart()
/**
* @param \Swoole\Server $server
*/
public function onManagerStart(\Swoole\Server $server)
{
$this->runEvent(Constant::MANAGER_START, null, [$server]);
}
public function onManagerStop()
{
/**
* @param \Swoole\Server $server
*/
public function onManagerStop(\Swoole\Server $server)
{
$this->runEvent(Constant::MANAGER_STOP, null, [$server]);
}
Server/Protocol/Protocol.php
+1 -1
View File
@@ -12,7 +12,7 @@ abstract class Protocol
* @param $data
* @return array
*/
public function resolveProtocol($data)
public function resolveProtocol($data): array
{
$explode = explode("\r\n\r\n", $data);
Server/Protocol/WebSocket.php
+60 -61
View File
@@ -7,76 +7,75 @@ class WebSocket extends Protocol
{
//
public function decode($received): ?string
{
$decoded = null;
$buffer = $received;
$len = ord($buffer[1]) & 127;
if ($len === 126) {
$masks = substr($buffer, 4, 4);
$data = substr($buffer, 8);
} else {
if ($len === 127) {
$masks = substr($buffer, 10, 4);
$data = substr($buffer, 14);
} else {
$masks = substr($buffer, 2, 4);
$data = substr($buffer, 6);
}
}
for ($index = 0; $index < strlen($data); $index++) {
$decoded .= $data[$index] ^ $masks[$index % 4];
}
//
public function decode($received): ?string
{
$decoded = null;
$buffer = $received;
$len = ord($buffer[1]) & 127;
if ($len === 126) {
$masks = substr($buffer, 4, 4);
$data = substr($buffer, 8);
} else {
if ($len === 127) {
$masks = substr($buffer, 10, 4);
$data = substr($buffer, 14);
} else {
$masks = substr($buffer, 2, 4);
$data = substr($buffer, 6);
}
}
for ($index = 0; $index < strlen($data); $index++) {
$decoded .= $data[$index] ^ $masks[$index % 4];
}
return $decoded;
}
return $decoded;
}
const BINARY_TYPE_BLOB = "\x81";
const BINARY_TYPE_BLOB = "\x81";
public function encode($buffer): string
{
$len = strlen($buffer);
public function encode($buffer): string
{
$len = strlen($buffer);
$first_byte = self::BINARY_TYPE_BLOB;
$first_byte = self::BINARY_TYPE_BLOB;
if ($len <= 125) {
$encode_buffer = $first_byte . chr($len) . $buffer;
} else {
if ($len <= 65535) {
$encode_buffer = $first_byte . chr(126) . pack("n", $len) . $buffer;
} else {
//pack("xxxN", $len)pack函数只处理2的32次方大小的文件，实际上2的32次方已经4G了。
$encode_buffer = $first_byte . chr(127) . pack("xxxxN", $len) . $buffer;
}
}
if ($len <= 125) {
$encode_buffer = $first_byte . chr($len) . $buffer;
} else {
if ($len <= 65535) {
$encode_buffer = $first_byte . chr(126) . pack("n", $len) . $buffer;
} else {
//pack("xxxN", $len)pack函数只处理2的32次方大小的文件，实际上2的32次方已经4G了。
$encode_buffer = $first_byte . chr(127) . pack("xxxxN", $len) . $buffer;
}
}
return $encode_buffer;
}
return $encode_buffer;
}
/**
* @param $server
* @param $fd
* @param $data
*/
private function getWebSocketProtocol($data)
{
[$http_protocol, $body] = $this->resolveProtocol($data);
$key = base64_encode(sha1($http_protocol['Sec-WebSocket-Key'] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', TRUE));
$headers = [
'HTTP/1.1 101 Switching Protocols',
'Upgrade: websocket',
'Connection: Upgrade',
'Sec-WebSocket-Accept: ' . $key,
'Sec-WebSocket-Version: 13',
];
if (isset($http_protocol['Sec-WebSocket-Protocol'])) {
$headers[] = 'Sec-WebSocket-Protocol: ' . $http_protocol['Sec-WebSocket-Protocol'];
}
return implode("\r\n", $headers) . "\r\n\r\n";
}
/**
* @param $data
* @return string
*/
private function getWebSocketProtocol($data): string
{
[$http_protocol, $body] = $this->resolveProtocol($data);
$key = base64_encode(sha1($http_protocol['Sec-WebSocket-Key'] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', TRUE));
$headers = [
'HTTP/1.1 101 Switching Protocols',
'Upgrade: websocket',
'Connection: Upgrade',
'Sec-WebSocket-Accept: ' . $key,
'Sec-WebSocket-Version: 13',
];
if (isset($http_protocol['Sec-WebSocket-Protocol'])) {
$headers[] = 'Sec-WebSocket-Protocol: ' . $http_protocol['Sec-WebSocket-Protocol'];
}
return implode("\r\n", $headers) . "\r\n\r\n";
}
}
Server/ServerManager.php
+364
View File
@@ -0,0 +1,364 @@
<?php
namespace Server;
use Closure;
use ReflectionException;
use Server\SInterface\CustomProcess;
use Server\Task\ServerTask;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;
use Swoole\Http\Server as HServer;
use Swoole\Process;
use Swoole\Server;
use Swoole\WebSocket\Server as WServer;
/**
* Class ServerManager
* @package HttpServer\Service
*/
class ServerManager extends Abstracts\Server
{
public string $host = '';
public int $port = 0;
/** @var Server\Port[] */
public array $ports = [];
public int $mode = SWOOLE_TCP;
private Server|WServer|HServer|null $server = null;
private static ?ServerManager $BASEServerListener = null;
/**
* @return static
*/
public static function getContext(): static
{
if (!(static::$BASEServerListener)) {
static::$BASEServerListener = new ServerManager();
}
return static::$BASEServerListener;
}
/**
* @return Server|WServer|HServer|null
*/
public function getServer(): Server|WServer|HServer|null
{
return $this->server;
}
/**
* @param string $type
* @param string $host
* @param int $port
* @param int $mode
* @param array $settings
* @throws NotFindClassException
* @throws ReflectionException
*/
public function addListener(string $type, string $host, int $port, int $mode, array $settings = [])
{
if (!$this->server) {
$this->createBaseServer($type, $host, $port, $mode, $settings);
} else {
if (!isset($settings['settings'])) {
$settings['settings'] = [];
}
$this->addNewListener($type, $host, $port, $mode, $settings);
}
}
/**
* @throws NotFindClassException
* @throws ReflectionException
*/
public function start(): void
{
$context = ServerManager::getContext();
$configs = require_once 'server.php';
foreach ($this->sortService($configs['server']['ports']) as $config) {
$this->startListenerHandler($context, $config);
}
$this->addServerEventCallback($this->getSystemEvents($configs));
$context->server->start();
}
/**
* @param string|CustomProcess $customProcess
* @param null $redirect_stdin_and_stdout
* @param int|null $pipe_type
* @param bool $enable_coroutine
*/
public function addProcess(string|CustomProcess $customProcess, $redirect_stdin_and_stdout = null, ?int $pipe_type = SOCK_DGRAM, bool $enable_coroutine = true)
{
if (is_string($customProcess)) {
$implements = class_implements($customProcess);
if (!in_array(CustomProcess::class, $implements)) {
trigger_error('custom process must implement ' . CustomProcess::class);
}
$customProcess = new $customProcess($this->server);
}
/** @var Process $process */
$this->server->addProcess(new Process(function (Process $soloProcess) use ($customProcess) {
$soloProcess->name($customProcess->getProcessName($soloProcess));
$customProcess->onHandler($soloProcess);
},
$redirect_stdin_and_stdout, $pipe_type, $enable_coroutine));
}
/**
* @param array $ports
* @return array
*/
private function sortService(array $ports): array
{
$array = [];
foreach ($ports as $port) {
if ($port['type'] == Constant::SERVER_TYPE_WEBSOCKET) {
array_unshift($array, $port);
} else if ($port['type'] == Constant::SERVER_TYPE_HTTP) {
if (!empty($array) && $array[0]['type'] == Constant::SERVER_TYPE_WEBSOCKET) {
$array[] = $port;
} else {
array_unshift($array, $port);
}
} else {
$array[] = $port;
}
}
return $array;
}
/**
* @param array $configs
* @return array
*/
private function getSystemEvents(array $configs): array
{
return array_intersect_key($configs['server']['events'] ?? [], [
Constant::PIPE_MESSAGE => '',
Constant::SHUTDOWN => '',
Constant::WORKER_START => '',
Constant::WORKER_ERROR => '',
Constant::WORKER_EXIT => '',
Constant::WORKER_STOP => '',
Constant::MANAGER_START => '',
Constant::MANAGER_STOP => '',
Constant::BEFORE_RELOAD => '',
Constant::AFTER_RELOAD => '',
Constant::DISCONNECT => '',
Constant::START => '',
]);
}
/**
* @param ServerManager $context
* @param array $config
* @throws NotFindClassException
* @throws ReflectionException
*/
private function startListenerHandler(ServerManager $context, array $config)
{
if ($this->server) {
$context->addNewListener($config['type'], $config['host'], $config['port'], $config['mode'], $config);
} else {
$config['settings'] = array_merge($configs['settings'] ?? [], $config['settings'] ?? []);
$config['events'] = array_merge($configs['events'] ?? [], $config['events'] ?? []);
$context->createBaseServer($config['type'], $config['host'], $config['port'], $config['mode'], $config);
}
}
/**
* @param string $type
* @param string $host
* @param int $port
* @param int $mode
* @param array $settings
* @throws ReflectionException
* @throws NotFindClassException
*/
private function addNewListener(string $type, string $host, int $port, int $mode, array $settings = [])
{
switch ($type) {
case Constant::SERVER_TYPE_TCP:
$this->ports[$port] = TCPServerListener::instance($this->server, $host, $port, $mode, $settings);
break;
case Constant::SERVER_TYPE_UDP:
$this->ports[$port] = UDPServerListener::instance($this->server, $host, $port, $mode, $settings);
break;
case Constant::SERVER_TYPE_HTTP:
$this->ports[$port] = HTTPServerListener::instance($this->server, $host, $port, $mode, $settings);
break;
case Constant::SERVER_TYPE_WEBSOCKET:
$this->ports[$port] = WebSocketServerListener::instance($this->server, $host, $port, $mode, $settings);
break;
}
}
/**
* @param int $port
* @param string $event
* @return Closure|array|null
*/
public function getPortCallback(int $port, string $event): Closure|array|null
{
/** @var Server\Port $_port */
$_port = $this->ports[$port] ?? null;
if (is_null($_port)) {
return null;
}
return $_port->getCallback($event);
}
/**
* @param string $type
* @param string $host
* @param int $port
* @param int $mode
* @param array $settings
* @throws NotFindClassException
* @throws ReflectionException
*/
private function createBaseServer(string $type, string $host, int $port, int $mode, array $settings = [])
{
$match = match ($type) {
Constant::SERVER_TYPE_BASE, Constant::SERVER_TYPE_TCP,
Constant::SERVER_TYPE_UDP => Server::class,
Constant::SERVER_TYPE_HTTP => HServer::class,
Constant::SERVER_TYPE_WEBSOCKET => WServer::class
};
$this->server = new $match($host, $port, SWOOLE_PROCESS, $mode);
$this->server->set($settings['settings']);
$this->addDefaultListener($type, $settings);
}
/**
* @param string $type
* @param array $settings
* @throws NotFindClassException
* @throws ReflectionException
*/
private function addDefaultListener(string $type, array $settings): void
{
$this->addServerEventCallback($settings['events']);
if (($this->server->setting['task_worker_num'] ?? 0) > 0) {
$this->addTaskListener($settings['events']);
}
if ($type === Constant::SERVER_TYPE_WEBSOCKET) {
$reflect = $this->getNewInstance(WebSocketServerListener::class);
$this->server->on('handshake', [$reflect, 'onHandshake']);
$this->server->on('message', [$reflect, 'onMessage']);
$this->server->on('close', [$reflect, 'onClose']);
$reflect->setEvents(Constant::HANDSHAKE, $settings['events'][Constant::HANDSHAKE] ?? null);
$reflect->setEvents(Constant::MESSAGE, $settings['events'][Constant::MESSAGE] ?? null);
$reflect->setEvents(Constant::CONNECT, $settings['events'][Constant::CONNECT] ?? null);
$reflect->setEvents(Constant::CLOSE, $settings['events'][Constant::CLOSE] ?? null);
} else if ($type === Constant::SERVER_TYPE_UDP) {
$reflect = $this->getNewInstance(UDPServerListener::class);
$this->server->on('packet', [$reflect, 'onPacket']);
$reflect->setEvents(Constant::PACKET, $settings['events'][Constant::PACKET] ?? null);
} else if ($type === Constant::SERVER_TYPE_HTTP) {
$reflect = $this->getNewInstance(HTTPServerListener::class);
$this->server->on('request', [$reflect, 'onRequest']);
$this->addCloseOrDisconnect($reflect, $settings);
} else {
$reflect = $this->getNewInstance(TCPServerListener::class);
$this->server->on('connect', [$reflect, 'onConnect']);
$this->server->on('receive', [$reflect, 'onReceive']);
$this->addCloseOrDisconnect($reflect, $settings);
}
}
/**
* @param string $class
* @return object
* @throws NotFindClassException
* @throws ReflectionException
*/
private function getNewInstance(string $class): object
{
return Snowflake::getDi()->getReflect($class)?->newInstance();
}
/**
* @param $reflect
* @param $settings
*/
private function addCloseOrDisconnect($reflect, $settings): void
{
if (swoole_version() >= '4.7.0') {
$this->server->on('disconnect', [$reflect, 'onDisconnect']);
$reflect->setEvents(Constant::DISCONNECT, $settings['events'][Constant::DISCONNECT] ?? null);
} else {
$this->server->on('close', [$reflect, 'onClose']);
$reflect->setEvents(Constant::CLOSE, $settings['events'][Constant::CLOSE] ?? null);
}
}
/**
* @param array $events
*/
private function addServerEventCallback(array $events): void
{
if (count($events) < 1) {
return;
}
foreach ($events as $event_type => $callback) {
if ($this->server->getCallback($event_type) !== null) {
continue;
}
$this->server->on($event_type, $callback);
}
}
/**
* @param array $events
* @throws NotFindClassException
* @throws ReflectionException
*/
private function addTaskListener(array $events = []): void
{
$task_use_object = $this->server->setting['task_object'] ?? $this->server->setting['task_use_object'] ?? false;
$reflect = Snowflake::getDi()->getReflect(ServerTask::class)?->newInstance();
if ($task_use_object || $this->server->setting['task_enable_coroutine']) {
$this->server->on('task', $events[Constant::TASK] ?? [$reflect, 'onCoroutineTask']);
} else {
$this->server->on('task', $events[Constant::TASK] ?? [$reflect, 'onTask']);
}
$this->server->on('finish', $events[Constant::FINISH] ?? [$reflect, 'onFinish']);
}
}
$context = ServerManager::getContext();
$context->start();
Server/TCPServerListener.php
+80 -53
View File
@@ -2,7 +2,9 @@
namespace Server;
use HttpServer\Route\Router;
use Exception;
use ReflectionException;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;
use Swoole\Server;
@@ -11,66 +13,91 @@ use Swoole\Server;
* Class TCPServerListener
* @package HttpServer\Service
*/
class TCPServerListener
class TCPServerListener extends Abstracts\Server
{
use ListenerHelper;
use ListenerHelper;
protected static mixed $_tcp;
/**
* UDPServerListener constructor.
* @param Server $server
* @param string $host
* @param int $port
* @param int $mode
* @param array|null $settings
* @throws \ReflectionException
*/
public static function instance(Server $server, string $host, int $port, int $mode, ?array $settings = [])
{
if (!in_array($mode, [SWOOLE_TCP, SWOOLE_TCP6])) {
trigger_error('Port mode ' . $host . '::' . $port . ' must is tcp listener type.');
}
$reflect = Snowflake::getDi()->getReflect(static::class)?->newInstance();
static::$_tcp = $server->addlistener($host, $port, $mode);
static::$_tcp->set($settings['settings'] ?? []);
static::$_tcp->on('receive', self::callback(Constant::RECEIVE, $settings['events'], [$reflect, 'onReceive']));
static::$_tcp->on('connect', static::callback(Constant::CONNECT, $settings['events'], [$reflect, 'onConnect']));
static::$_tcp->on('close', static::callback(Constant::CLOSE, $settings['events'], [$reflect, 'onClose']));
}
protected static mixed $_tcp;
/**
* @param Server $server
* @param int $fd
*/
public function onConnect(Server $server, int $fd)
{
var_dump(__FILE__ . ':' . __LINE__);
}
/**
* UDPServerListener constructor.
* @param Server $server
* @param string $host
* @param int $port
* @param int $mode
* @param array|null $settings
* @return Server\Port
* @throws NotFindClassException
* @throws ReflectionException
*/
public static function instance(Server $server, string $host, int $port, int $mode, ?array $settings = []): Server\Port
{
if (!in_array($mode, [SWOOLE_TCP, SWOOLE_TCP6])) {
trigger_error('Port mode ' . $host . '::' . $port . ' must is tcp listener type.');
}
/** @var static $reflect */
$reflect = Snowflake::getDi()->getReflect(static::class)?->newInstance();
static::$_tcp = $server->addlistener($host, $port, $mode);
static::$_tcp->set($settings['settings'] ?? []);
static::$_tcp->on('receive', [$reflect, 'onReceive']);
static::$_tcp->on('connect', [$reflect, 'onConnect']);
if (swoole_version() >= '4.7.0') {
static::$_tcp->on('disconnect', [$reflect, 'onDisconnect']);
$reflect->setEvents(Constant::DISCONNECT, $settings['events'][Constant::DISCONNECT] ?? null);
} else {
static::$_tcp->on('close', [$reflect, 'onClose']);
$reflect->setEvents(Constant::CLOSE, $settings['events'][Constant::CLOSE] ?? null);
}
$reflect->setEvents(Constant::RECEIVE, $settings['events'][Constant::RECEIVE] ?? null);
$reflect->setEvents(Constant::CONNECT, $settings['events'][Constant::CONNECT] ?? null);
return static::$_tcp;
}
/**
* @param Server $server
* @param int $fd
* @param int $reactor_id
* @param string $data
*/
public function onReceive(Server $server, int $fd, int $reactor_id, string $data)
{
var_dump($data);
$server->send($fd, $data);
}
/**
* @param Server $server
* @param int $fd
* @param int|null $reactor_id
*/
public function onDisconnect(Server $server, int $fd, ?int $reactor_id = null)
{
$this->runEvent(Constant::HANDSHAKE, null, [$server, $fd, $reactor_id]);
}
/**
* @param Server $server
* @param int $fd
*/
public function onClose(Server $server, int $fd)
{
}
/**
* @param Server $server
* @param int $fd
*/
public function onConnect(Server $server, int $fd)
{
$this->runEvent(Constant::CONNECT, null, [$server, $fd]);
}
/**
* @param Server $server
* @param int $fd
* @param int $reactor_id
* @param string $data
*/
public function onReceive(Server $server, int $fd, int $reactor_id, string $data)
{
$this->runEvent(Constant::RECEIVE, null, [$server, $fd, $reactor_id, $data]);
}
/**
* @param Server $server
* @param int $fd
*/
public function onClose(Server $server, int $fd)
{
$this->runEvent(Constant::CLOSE, null, [$server, $fd]);
}
}
Server/Task/ServerTask.php
+9 -2
View File
@@ -4,10 +4,17 @@
namespace Server\Task;
use ReflectionException;
use Server\SInterface\TaskExecute;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;
use Swoole\Server;
/**
* Class ServerTask
* @package Server\Task
*/
class ServerTask
{
@@ -49,8 +56,8 @@ class ServerTask
/**
* @param $data
* @return null
* @throws \ReflectionException
* @throws \Snowflake\Exception\NotFindClassException
* @throws ReflectionException
* @throws NotFindClassException
*/
private function resolve($data)
{
Server/UDPServerListener.php
+43 -28
View File
@@ -2,6 +2,9 @@
namespace Server;
use Exception;
use ReflectionException;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;
use Swoole\Server;
@@ -10,41 +13,53 @@ use Swoole\Server;
* Class UDPServerListener
* @package HttpServer\Service
*/
class UDPServerListener
class UDPServerListener extends Abstracts\Server
{
protected static mixed $_udp;
protected static mixed $_udp;
use ListenerHelper;
use ListenerHelper;
/**
* @param Server $server
* @param string $host
* @param int $port
* @param int $mode
* @param array|null $settings
*/
public static function instance(Server $server, string $host, int $port, int $mode, ?array $settings = [])
{
if (!in_array($mode, [SWOOLE_UDP, SWOOLE_UDP6])) {
trigger_error('Port mode ' . $host . '::' . $port . ' must is udp listener type.');
}
static::$_udp = $server->addlistener($host, $port, $mode);
static::$_udp->set($settings['settings'] ?? []);
static::$_udp->on('packet', static::callback(Constant::PACKET, $settings['events'], [new static(), 'onPacket']));
}
/**
* @param Server $server
* @param string $host
* @param int $port
* @param int $mode
* @param array|null $settings
* @return Server\Port
* @throws NotFindClassException
* @throws ReflectionException
* @throws Exception
*/
public static function instance(Server $server, string $host, int $port, int $mode, ?array $settings = []): Server\Port
{
if (!in_array($mode, [SWOOLE_UDP, SWOOLE_UDP6])) {
trigger_error('Port mode ' . $host . '::' . $port . ' must is udp listener type.');
}
/** @var static $reflect */
$reflect = Snowflake::getDi()->getReflect(static::class)->newInstance();
static::$_udp = $server->addlistener($host, $port, $mode);
static::$_udp->set($settings['settings'] ?? []);
static::$_udp->on('packet', [$reflect, 'onPacket']);
$reflect->setEvents(Constant::PACKET, $settings['events'][Constant::PACKET] ?? null);
return static::$_udp;
}
/**
* @param Server $server
* @param string $data
* @param array $clientInfo
*/
public function onPacket(Server $server, string $data, array $clientInfo)
{
$server->sendto($clientInfo['address'], $clientInfo['port'], $data);
}
/**
* @param Server $server
* @param string $data
* @param array $clientInfo
*/
public function onPacket(Server $server, string $data, array $clientInfo)
{
$this->runEvent(Constant::MESSAGE, null, [$server, $data, $clientInfo]);
}
}
Server/WebSocketServerListener.php
+102 -76
View File
@@ -3,6 +3,8 @@
namespace Server;
use Exception;
use ReflectionException;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;
use Swoole\Http\Request;
use Swoole\Http\Response;
@@ -14,93 +16,117 @@ use Swoole\WebSocket\Frame;
* Class WebSocketServerListener
* @package HttpServer\Service
*/
class WebSocketServerListener
class WebSocketServerListener extends Abstracts\Server
{
protected static Server\Port $_http;
protected static Server\Port $_http;
use ListenerHelper;
use ListenerHelper;
/**
* UDPServerListener constructor.
* @param Server|\Swoole\WebSocket\Server|\Swoole\Http\Server $server
* @param string $host
* @param int $port
* @param int $mode
* @param array|null $settings
*/
public static function instance(mixed $server, string $host, int $port, int $mode, ?array $settings = [])
{
if (!in_array($mode, [SWOOLE_TCP, SWOOLE_TCP6])) {
trigger_error('Port mode ' . $host . '::' . $port . ' must is udp listener type.');
}
static::$_http = $server->addlistener($host, $port, $mode);
static::$_http->set($settings['settings'] ?? []);
static::$_http->on('connect', static::callback(Constant::CONNECT, $settings['events'], [new static(), 'onConnect']));
static::$_http->on('handshake', static::callback(Constant::HANDSHAKE, $settings['events'], [new static(), 'onHandshake']));
static::$_http->on('message', static::callback(Constant::MESSAGE, $settings['events'], [new static(), 'onMessage']));
static::$_http->on('close', static::callback(Constant::CLOSE, $settings['events'], [new static(), 'onClose']));
}
/**
* @param mixed $server
* @param string $host
* @param int $port
* @param int $mode
* @param array|null $settings
* @return Server\Port
* @throws NotFindClassException
* @throws ReflectionException
* @throws Exception
*/
public static function instance(mixed $server, string $host, int $port, int $mode, ?array $settings = []): Server\Port
{
if (!in_array($mode, [SWOOLE_TCP, SWOOLE_TCP6])) {
trigger_error('Port mode ' . $host . '::' . $port . ' must is udp listener type.');
}
/** @var static $reflect */
$reflect = Snowflake::getDi()->getReflect(static::class)?->newInstance();
static::$_http = $server->addlistener($host, $port, $mode);
static::$_http->set($settings['settings'] ?? []);
static::$_http->on('connect', [$reflect, 'onConnect']);
static::$_http->on('handshake', [$reflect, 'onHandshake']);
static::$_http->on('message', [$reflect, 'onMessage']);
static::$_http->on('close', [$reflect, 'onClose']);
$reflect->setEvents(Constant::HANDSHAKE, $settings['events'][Constant::CONNECT] ?? null);
$reflect->setEvents(Constant::MESSAGE, $settings['events'][Constant::HANDSHAKE] ?? null);
$reflect->setEvents(Constant::CONNECT, $settings['events'][Constant::MESSAGE] ?? null);
$reflect->setEvents(Constant::CLOSE, $settings['events'][Constant::CLOSE] ?? null);
return static::$_http;
}
/**
* @param Request $request
* @param Response $response
* @throws Exception
*/
public function onHandshake(Request $request, Response $response)
{
/** @var \Swoole\WebSocket\Server $server */
$secWebSocketKey = $request->header['sec-websocket-key'];
$patten = '#^[+/0-9A-Za-z]{21}[AQgw]==$#';
if (0 === preg_match($patten, $secWebSocketKey) || 16 !== strlen(base64_decode($secWebSocketKey))) {
throw new Exception('protocol error.', 500);
}
$key = base64_encode(sha1($request->header['sec-websocket-key'] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', TRUE));
$headers = [
'Upgrade' => 'websocket',
'Connection' => 'Upgrade',
'Sec-websocket-Accept' => $key,
'Sec-websocket-Version' => '13',
];
if (isset($request->header['sec-websocket-protocol'])) {
$headers['Sec-websocket-Protocol'] = $request->header['sec-websocket-protocol'];
}
foreach ($headers as $key => $val) {
$response->setHeader($key, $val);
}
$response->setStatusCode(101);
$response->end();
}
/**
* @param Request $request
* @param Response $response
* @throws Exception
*/
public function onHandshake(Request $request, Response $response)
{
/** @var \Swoole\WebSocket\Server $server */
$secWebSocketKey = $request->header['sec-websocket-key'];
$patten = '#^[+/0-9A-Za-z]{21}[AQgw]==$#';
if (0 === preg_match($patten, $secWebSocketKey) || 16 !== strlen(base64_decode($secWebSocketKey))) {
throw new Exception('protocol error.', 500);
}
$key = base64_encode(sha1($request->header['sec-websocket-key'] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', TRUE));
$headers = [
'Upgrade' => 'websocket',
'Connection' => 'Upgrade',
'Sec-websocket-Accept' => $key,
'Sec-websocket-Version' => '13',
];
if (isset($request->header['sec-websocket-protocol'])) {
$headers['Sec-websocket-Protocol'] = $request->header['sec-websocket-protocol'];
}
foreach ($headers as $key => $val) {
$response->setHeader($key, $val);
}
$this->runEvent(Constant::HANDSHAKE, fn() => $this->disconnect($request, $response), [$request, $response]);
}
/**
* @param Server $server
* @param int $fd
*/
public function onConnect(Server $server, int $fd)
{
var_dump(__FILE__ . ':' . __LINE__);
$server->confirm($fd);
}
/**
* @param $request
* @param $response
*/
public function disconnect($request, $response)
{
$response->setStatusCode(502);
$response->end();
}
/**
* @param \Swoole\WebSocket\Server|Server $server
* @param Frame $frame
*/
public function onMessage(\Swoole\WebSocket\Server|Server $server, Frame $frame)
{
}
/**
* @param Server $server
* @param int $fd
*/
public function onConnect(Server $server, int $fd)
{
$this->runEvent(Constant::MESSAGE, fn() => $server->confirm($fd), [$server, $fd]);
}
/**
* @param Server $server
* @param int $fd
*/
public function onClose(Server $server, int $fd)
{
var_dump($server->getClientInfo($fd));
}
/**
* @param \Swoole\WebSocket\Server $server
* @param Frame $frame
*/
public function onMessage(\Swoole\WebSocket\Server $server, Frame $frame)
{
$this->runEvent(Constant::MESSAGE, fn() => $server->push($frame->fd, '.'), [$server, $frame]);
}
/**
* @param Server $server
* @param int $fd
*/
public function onClose(Server $server, int $fd)
{
$this->runEvent(Constant::MESSAGE, fn() => $server->confirm($fd), [$server, $fd]);
}
}
Server/Worker/ServerWorker.php
+40 -4
View File
@@ -2,19 +2,55 @@
namespace Server\Worker;
use Exception;
use Server\Constant;
use Snowflake\Abstracts\Config;
use Snowflake\Event;
use Snowflake\Runtime;
use Snowflake\Snowflake;
use Swoole\Server;
class ServerWorker
/**
* Class ServerWorker
* @package Server\Worker
*/
class ServerWorker extends \Server\Abstracts\Server
{
/**
* @param Server $server
* @param int $workerId
* @throws Exception
*/
public function onWorkerStart(Server $server, int $workerId)
{
$annotation = Snowflake::app()->getAnnotation();
$annotation->read(APP_PATH);
$this->runEvent(Constant::WORKER_START, null, [$server, $workerId]);
if ($workerId >= $server->setting['worker_num'] + 1) {
$annotation->runtime(CONTROLLER_PATH);
}
Event::trigger(Event::SERVER_ON_WORKER_START, [$server, $workerId]);
name($server->worker_pid, 'Worker.' . $workerId);
}
/**
* @param $worker_id
* @throws Exception
*/
private function setConfigs($worker_id)
{
putenv('state=start');
putenv('worker=' . $worker_id);
$serialize = file_get_contents(storage(Runtime::CONFIG_NAME));
if (empty($serialize)) {
return;
}
Config::sets(unserialize($serialize));
}
@@ -24,7 +60,7 @@ class ServerWorker
*/
public function onWorkerStop(Server $server, int $workerId)
{
$this->runEvent(Constant::WORKER_STOP, null, [$server, $workerId]);
}
@@ -34,7 +70,7 @@ class ServerWorker
*/
public function onWorkerExit(Server $server, int $workerId)
{
$this->runEvent(Constant::WORKER_EXIT, null, [$server, $workerId]);
}
@@ -47,7 +83,7 @@ class ServerWorker
*/
public function onWorkerError(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal)
{
$this->runEvent(Constant::WORKER_ERROR, null, [$server, $worker_id, $worker_pid, $exit_code, $signal]);
}
}
System/Event.php
+3 -3
View File
@@ -218,7 +218,7 @@ class Event extends BaseObject
* @return bool
* @throws Exception
*/
public static function trigger($name, $parameter = null, $handler = null, $is_remove = false): bool
public static function trigger($name, $parameter = null, $handler = null, bool $is_remove = false): bool
{
$events = static::get($name, $handler);
if (empty($events)) {
@@ -256,10 +256,10 @@ class Event extends BaseObject
/**
* @param $defaultParameter
* @param $parameter
* @param array $parameter
* @return array
*/
private static function mergeParams($defaultParameter, $parameter = []): array
private static function mergeParams($defaultParameter, array $parameter = []): array
{
if (empty($defaultParameter)) {
$defaultParameter = $parameter;