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-04-23 03:25:03 +08:00
parent 0c5e05b6a2
commit a5a607495f
11 changed files with 261 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HttpServer/Events/OnClose.php
+25 -24
View File
@@ -19,37 +19,38 @@ class OnClose extends Callback
{
/**
* @param Server $server
* @param int $fd
* @throws Exception
*/
public function onHandler(Server $server, int $fd)
{
try {
/**
* @param Server $server
* @param int $fd
* @throws Exception
*/
public function onHandler(Server $server, int $fd)
{
try {
defer(function () {
fire(Event::SYSTEM_RESOURCE_RELEASES);
logger_insert();
});
$clientInfo = $server->getClientInfo($fd);
$event = Snowflake::app()->getEvent();
if (!$event->exists(($name = $this->getName($clientInfo)))) {
return;
}
return;
}
$event->trigger($name, [$server, $fd]);
} catch (\Throwable $exception) {
$this->addError($exception, 'throwable');
} finally {
fire(Event::SYSTEM_RESOURCE_RELEASES);
logger_insert();
}
}
$this->addError($exception, 'throwable');
}
}
/**
* @param $server_port
* @return string
*/
private function getName($server_port): string
{
return 'listen ' . $server_port['server_port'] . ' ' . Event::SERVER_CLIENT_CLOSE;
}
/**
* @param $server_port
* @return string
*/
private function getName($server_port): string
{
return 'listen ' . $server_port['server_port'] . ' ' . Event::SERVER_CLIENT_CLOSE;
}
}
HttpServer/Events/OnConnect.php
+5 -2
View File
@@ -26,6 +26,11 @@ class OnConnect extends Callback
*/
public function onHandler(Server $server, int $fd, int $reactorId)
{
defer(function () {
fire(Event::SYSTEM_RESOURCE_RELEASES);
logger_insert();
});
$event = Snowflake::app()->getEvent();
$clientInfo = $server->getClientInfo($fd);
@@ -37,8 +42,6 @@ class OnConnect extends Callback
} catch (\Throwable $throwable) {
$this->addError($throwable, 'connect');
}
fire(Event::SYSTEM_RESOURCE_RELEASES);
logger_insert();
}
HttpServer/Events/OnFinish.php
+18 -14
View File
@@ -18,19 +18,23 @@ class OnFinish extends Callback
{
/**
* @param Server $server
* @param $task_id
* @param $data
* @throws Exception
*/
public function onHandler(Server $server, $task_id, $data)
{
try {
fire(Event::TASK_FINISH, [$task_id, $data]);
} catch (\Throwable $exception) {
$this->addError($exception, 'task');
}
}
/**
* @param Server $server
* @param $task_id
* @param $data
* @throws Exception
*/
public function onHandler(Server $server, $task_id, $data)
{
try {
defer(function () {
fire(Event::SYSTEM_RESOURCE_RELEASES);
logger_insert();
});
fire(Event::TASK_FINISH, [$task_id, $data]);
} catch (\Throwable $exception) {
$this->addError($exception, 'task');
}
}
}
HttpServer/Events/OnHandshake.php
+108 -107
View File
@@ -29,131 +29,132 @@ class OnHandshake extends Callback
{
/**
* @param $request
* @param $response
* @throws Exception
*/
private function resolveParse($request, $response)
{
/** @var 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->header($key, $val);
}
}
/**
* @param $request
* @param $response
* @throws Exception
*/
private function resolveParse($request, $response)
{
/** @var 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->header($key, $val);
}
}
/**
* @param SResponse $response
* @param int $code
* @return false
*/
private function disconnect(SResponse $response, $code = 500): bool
{
$server = Snowflake::getWebSocket();
if (!$server->exist($response->fd)) {
return false;
}
$response->status($code);
$response->end();
return false;
}
/**
* @param SResponse $response
* @param int $code
* @return false
*/
private function disconnect(SResponse $response, $code = 500): bool
{
$server = Snowflake::getWebSocket();
if (!$server->exist($response->fd)) {
return false;
}
$response->status($code);
$response->end();
return false;
}
/**
* @param $response
* @param int $code
* @return false
*/
private function connect($response, $code = 101): bool
{
$response->status($code);
$response->end();
return false;
}
/**
* @param $response
* @param int $code
* @return false
*/
private function connect($response, $code = 101): bool
{
$response->status($code);
$response->end();
return false;
}
/**
* @param SRequest $request
* @param SResponse $response
* @return void
* @throws Exception
*/
public function onHandler(SRequest $request, SResponse $response): void
{
try {
$this->execute($request, $response);
} catch (\Throwable $exception) {
$this->addError($exception, 'throwable');
$response->status(500);
$response->end($exception->getMessage());
} finally {
fire(Event::SYSTEM_RESOURCE_CLEAN);
logger_insert();
}
}
/**
* @param SRequest $request
* @param SResponse $response
* @return void
* @throws Exception
*/
public function onHandler(SRequest $request, SResponse $response): void
{
try {
defer(function () {
fire(Event::SYSTEM_RESOURCE_CLEAN);
logger_insert();
});
$this->execute($request, $response);
} catch (\Throwable $exception) {
$this->addError($exception, 'throwable');
$response->status(500);
$response->end($exception->getMessage());
}
}
/**
* @param SRequest $request
* @param SResponse $response
* @return mixed
* @throws Exception
*/
private function execute(SRequest $request, SResponse $response): mixed
{
$this->resolveParse($request, $response);
/**
* @param SRequest $request
* @param SResponse $response
* @return mixed
* @throws Exception
*/
private function execute(SRequest $request, SResponse $response): mixed
{
$this->resolveParse($request, $response);
$router = Snowflake::app()->getRouter();
$router = Snowflake::app()->getRouter();
[$sRequest, $sResponse] = $this->sRequest($request, $response);
[$sRequest, $sResponse] = $this->sRequest($request, $response);
if (($node = $router->find_path($sRequest)) !== null) {
return $node->dispatch($sRequest, $sResponse);
}
return $this->disconnect($response, 404);
}
if (($node = $router->find_path($sRequest)) !== null) {
return $node->dispatch($sRequest, $sResponse);
}
return $this->disconnect($response, 404);
}
/**
* @param $request
* @param SResponse $response
* @return array
* @throws NotFindClassException
* @throws ReflectionException
*/
private function sRequest($request, SResponse $response): array
{
/** @var Request $sRequest */
$sRequest = Request::create($request);
$sRequest->uri = '/' . Socket::HANDSHAKE . '::event';
/**
* @param $request
* @param SResponse $response
* @return array
* @throws NotFindClassException
* @throws ReflectionException
*/
private function sRequest($request, SResponse $response): array
{
/** @var Request $sRequest */
$sRequest = Request::create($request);
$sRequest->uri = '/' . Socket::HANDSHAKE . '::event';
$sRequest->headers = new HttpHeaders(ArrayAccess::merge($request->server, $request->header));
$sRequest->headers = new HttpHeaders(ArrayAccess::merge($request->server, $request->header));
$sRequest->headers->replace('request_method', 'sw::socket');
$sRequest->headers->replace('request_uri', $sRequest->uri);
$sRequest->headers->replace('request_method', 'sw::socket');
$sRequest->headers->replace('request_uri', $sRequest->uri);
$sRequest->params = new HttpParams([], $request->get, []);
$sRequest->params = new HttpParams([], $request->get, []);
$sRequest->parseUri();
$sRequest->parseUri();
return [$sRequest, Response::create($response)];
}
return [$sRequest, Response::create($response)];
}
}
HttpServer/Events/OnMessage.php
+5 -4
View File
@@ -38,7 +38,11 @@ class OnMessage extends Callback
public function onHandler(Server $server, Frame $frame)
{
try {
if ($frame->opcode === 0x08) {
defer(function () {
fire(Event::SYSTEM_RESOURCE_CLEAN);
logger_insert();
});
if ($frame->opcode === 0x08) {
return;
}
$clientInfo = $server->getClientInfo($frame->fd);
@@ -51,9 +55,6 @@ class OnMessage extends Callback
} catch (\Throwable $exception) {
$this->addError($exception, 'websocket');
$server->send($frame->fd, $exception->getMessage());
} finally {
fire(Event::SYSTEM_RESOURCE_RELEASES);
logger_insert();
}
}
HttpServer/Events/OnPacket.php
+1 -1
View File
@@ -36,7 +36,7 @@ class OnPacket extends Callback
{
[$host, $port] = [$clientInfo['address'], $clientInfo['port']];
try {
\Swoole\Coroutine\defer(function () {
defer(function () {
fire(Event::SYSTEM_RESOURCE_RELEASES);
\logger_insert();
});
HttpServer/Events/OnPipeMessage.php
+7 -6
View File
@@ -26,14 +26,15 @@ class OnPipeMessage extends Callback
*/
public function onHandler(Server $server, int $src_worker_id, $swollen_universalize)
{
go(function () use ($server, $src_worker_id, $swollen_universalize) {
match ($swollen_universalize['action'] ?? null) {
'kafka' => $this->onKafkaWorker($swollen_universalize),
'crontab' => $this->onCrontabWorker($swollen_universalize),
default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize)
};
defer(function () {
fire(Event::SYSTEM_RESOURCE_RELEASES);
logger_insert();
});
match ($swollen_universalize['action'] ?? null) {
'kafka' => $this->onKafkaWorker($swollen_universalize),
'crontab' => $this->onCrontabWorker($swollen_universalize),
default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize)
};
}
HttpServer/Events/OnReceive.php
+1 -1
View File
@@ -48,7 +48,7 @@ class OnReceive extends Callback
public function onHandler(Server $server, int $fd, int $reID, string $data): mixed
{
try {
\Swoole\Coroutine\defer(function (){
defer(function (){
fire(Event::SYSTEM_RESOURCE_RELEASES);
\logger_insert();
});
HttpServer/Events/OnRequest.php
+68 -67
View File
@@ -24,88 +24,89 @@ class OnRequest extends Callback
{
public Event $event;
public Logger $logger;
public Event $event;
public Logger $logger;
/**
* @throws Exception
*/
public function init()
{
$this->event = Snowflake::app()->getEvent();
$this->logger = Snowflake::app()->getLogger();
}
/**
* @throws Exception
*/
public function init()
{
$this->event = Snowflake::app()->getEvent();
$this->logger = Snowflake::app()->getLogger();
}
/**
* @param Request $request
* @param Response $response
* @return void
* @throws Exception
*/
public function onHandler(Request $request, Response $response): mixed
{
try {
/** @var HRequest $request */
[$request, $response] = OnRequest::createContext($request, $response);
/**
* @param Request $request
* @param Response $response
* @return void
* @throws Exception
*/
public function onHandler(Request $request, Response $response): mixed
{
try {
defer(function () {
$this->event->trigger(Event::SYSTEM_RESOURCE_RELEASES);
$this->logger->insert();
});
/** @var HRequest $request */
[$request, $response] = OnRequest::createContext($request, $response);
$this->event->dispatch(Event::EVENT_BEFORE_REQUEST, [$request]);
$this->event->dispatch(Event::EVENT_BEFORE_REQUEST, [$request]);
$result = $request->dispatch();
$result = $request->dispatch();
$this->event->dispatch(Event::EVENT_AFTER_REQUEST, [$request, $result]);
$this->event->dispatch(Event::EVENT_AFTER_REQUEST, [$request, $result]);
return $result;
} catch (ExitException | Error | \Throwable $exception) {
$this->addError($exception, 'throwable');
return $this->sendErrorMessage($request, $response, $exception);
} finally {
$this->event->trigger(Event::SYSTEM_RESOURCE_RELEASES);
$this->logger->insert();
}
}
return $result;
} catch (ExitException | Error | \Throwable $exception) {
$this->addError($exception, 'throwable');
return $this->sendErrorMessage($request, $response, $exception);
}
}
/**
* @param $request
* @param $response
* @return array
*/
public static function createContext($request, $response): array
{
return [HRequest::create($request), HResponse::create($response)];
}
/**
* @param $request
* @param $response
* @return array
*/
public static function createContext($request, $response): array
{
return [HRequest::create($request), HResponse::create($response)];
}
/**
* @param $sRequest
* @param $sResponse
* @param $exception
* @return bool|string
* @throws Exception
*/
protected function sendErrorMessage($sRequest, $sResponse, $exception): bool|string
{
$this->addError($exception, 'throwable');
if ($sResponse instanceof Response) {
[$sRequest, $sResponse] = [HRequest::create($sRequest), HResponse::create($sResponse)];
}
/**
* @param $sRequest
* @param $sResponse
* @param $exception
* @return bool|string
* @throws Exception
*/
protected function sendErrorMessage($sRequest, $sResponse, $exception): bool|string
{
$this->addError($exception, 'throwable');
if ($sResponse instanceof Response) {
[$sRequest, $sResponse] = [HRequest::create($sRequest), HResponse::create($sResponse)];
}
$this->event->dispatch(Event::EVENT_AFTER_REQUEST, [$sRequest, $exception]);
$this->event->dispatch(Event::EVENT_AFTER_REQUEST, [$sRequest, $exception]);
$headers = $sRequest->headers->get('access-control-request-headers');
$methods = $sRequest->headers->get('access-control-request-method');
$headers = $sRequest->headers->get('access-control-request-headers');
$methods = $sRequest->headers->get('access-control-request-method');
$sResponse->addHeader('Access-Control-Allow-Origin', '*');
$sResponse->addHeader('Access-Control-Allow-Headers', $headers);
$sResponse->addHeader('Access-Control-Request-Method', $methods);
$sResponse->addHeader('Access-Control-Allow-Origin', '*');
$sResponse->addHeader('Access-Control-Allow-Headers', $headers);
$sResponse->addHeader('Access-Control-Request-Method', $methods);
if (!($exception instanceof ExitException)) {
return $sResponse->send(\logger()->exception($exception), 200);
} else {
return $sResponse->send($exception->getMessage(), 200);
}
}
if (!($exception instanceof ExitException)) {
return $sResponse->send(\logger()->exception($exception), 200);
} else {
return $sResponse->send($exception->getMessage(), 200);
}
}
}
HttpServer/Events/OnTask.php
+5 -4
View File
@@ -102,7 +102,11 @@ class OnTask extends Callback
private function runTaskHandler($data): ?array
{
try {
$serialize = $this->before($data);
defer(function () {
fire(Event::SYSTEM_RESOURCE_CLEAN);
logger_insert();
});
$serialize = $this->before($data);
if ($serialize === null) {
throw new Exception('unpack error.');
}
@@ -121,9 +125,6 @@ class OnTask extends Callback
$this->addError($exception, 'Task');
return $finish;
} finally {
fire(Event::SYSTEM_RESOURCE_RELEASES);
logger_insert();
}
}
HttpServer/Events/OnWorkerError.php
+18 -18
View File
@@ -18,26 +18,26 @@ class OnWorkerError extends Callback
{
/**
* @param Server $server
* @param int $worker_id
* @param int $worker_pid
* @param int $exit_code
* @param int $signal
* @throws Exception
*/
public function onHandler(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal)
{
$event = Snowflake::app()->getEvent();
$event->trigger(Event::SERVER_WORKER_ERROR);
/**
* @param Server $server
* @param int $worker_id
* @param int $worker_pid
* @param int $exit_code
* @param int $signal
* @throws Exception
*/
public function onHandler(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal)
{
$event = Snowflake::app()->getEvent();
$event->trigger(Event::SERVER_WORKER_ERROR);
$message = sprintf('Worker#%d error stop. signal %d, exit_code %d',
$worker_id, $signal, $exit_code
);
$message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d',
$worker_id, $signal, $worker_pid, $exit_code
);
write($message, 'worker-exit');
write($message, 'worker-exit');
logger_insert();
}
logger_insert();
}
}