From a5a607495f38fdb72c485deb63967e1ba41e3faa Mon Sep 17 00:00:00 2001 From: "as2252258@163.com" Date: Fri, 23 Apr 2021 03:25:03 +0800 Subject: [PATCH] modify --- HttpServer/Events/OnClose.php | 49 +++---- HttpServer/Events/OnConnect.php | 7 +- HttpServer/Events/OnFinish.php | 32 +++-- HttpServer/Events/OnHandshake.php | 215 ++++++++++++++-------------- HttpServer/Events/OnMessage.php | 9 +- HttpServer/Events/OnPacket.php | 2 +- HttpServer/Events/OnPipeMessage.php | 13 +- HttpServer/Events/OnReceive.php | 2 +- HttpServer/Events/OnRequest.php | 135 ++++++++--------- HttpServer/Events/OnTask.php | 9 +- HttpServer/Events/OnWorkerError.php | 36 ++--- 11 files changed, 261 insertions(+), 248 deletions(-) diff --git a/HttpServer/Events/OnClose.php b/HttpServer/Events/OnClose.php index cdbe6ba2..cf9e0e9d 100644 --- a/HttpServer/Events/OnClose.php +++ b/HttpServer/Events/OnClose.php @@ -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; + } } diff --git a/HttpServer/Events/OnConnect.php b/HttpServer/Events/OnConnect.php index 09ffd501..0a85da5c 100644 --- a/HttpServer/Events/OnConnect.php +++ b/HttpServer/Events/OnConnect.php @@ -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(); } diff --git a/HttpServer/Events/OnFinish.php b/HttpServer/Events/OnFinish.php index 02f116ee..85e7825f 100644 --- a/HttpServer/Events/OnFinish.php +++ b/HttpServer/Events/OnFinish.php @@ -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'); + } + } } diff --git a/HttpServer/Events/OnHandshake.php b/HttpServer/Events/OnHandshake.php index c32a0e87..b1e7e876 100644 --- a/HttpServer/Events/OnHandshake.php +++ b/HttpServer/Events/OnHandshake.php @@ -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)]; + } } diff --git a/HttpServer/Events/OnMessage.php b/HttpServer/Events/OnMessage.php index c49b4848..55bc822f 100644 --- a/HttpServer/Events/OnMessage.php +++ b/HttpServer/Events/OnMessage.php @@ -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(); } } diff --git a/HttpServer/Events/OnPacket.php b/HttpServer/Events/OnPacket.php index 35b0ecc9..b83da590 100644 --- a/HttpServer/Events/OnPacket.php +++ b/HttpServer/Events/OnPacket.php @@ -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(); }); diff --git a/HttpServer/Events/OnPipeMessage.php b/HttpServer/Events/OnPipeMessage.php index 68d5da9d..91a1ed57 100644 --- a/HttpServer/Events/OnPipeMessage.php +++ b/HttpServer/Events/OnPipeMessage.php @@ -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) + }; } diff --git a/HttpServer/Events/OnReceive.php b/HttpServer/Events/OnReceive.php index 90863e96..c9b27131 100644 --- a/HttpServer/Events/OnReceive.php +++ b/HttpServer/Events/OnReceive.php @@ -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(); }); diff --git a/HttpServer/Events/OnRequest.php b/HttpServer/Events/OnRequest.php index 8e048364..712ba4e3 100644 --- a/HttpServer/Events/OnRequest.php +++ b/HttpServer/Events/OnRequest.php @@ -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); + } + } } diff --git a/HttpServer/Events/OnTask.php b/HttpServer/Events/OnTask.php index 6f2eb43b..dfe71485 100644 --- a/HttpServer/Events/OnTask.php +++ b/HttpServer/Events/OnTask.php @@ -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(); } } diff --git a/HttpServer/Events/OnWorkerError.php b/HttpServer/Events/OnWorkerError.php index 01ab34d4..ad813bc7 100644 --- a/HttpServer/Events/OnWorkerError.php +++ b/HttpServer/Events/OnWorkerError.php @@ -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(); + } }