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-21 17:55:34 +08:00
parent 503a188dd7
commit ad43351306
3 changed files with 95 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HttpServer/Route/Router.php
+5 -9
View File
@@ -545,21 +545,17 @@ class Router extends HttpService implements RouterInterface
*/ */
public function find_path(Request $request): ?Node public function find_path(Request $request): ?Node
{ {
// return $this->Branch_search($request);
$method = $request->getMethod(); $method = $request->getMethod();
$uri = $request->headers->get('request_uri', '/'); $uri = $request->headers->get('request_uri', '/');
if (!isset(static::$nodes[$method])) { $methods = static::$nodes[$method][$uri] ?? null;
return null; if (!is_null($methods[$uri])) {
}
$methods = static::$nodes[$method];
if (isset($methods[$uri])) {
return $methods[$uri]; return $methods[$uri];
} }
if (!$request->isOption || !isset($methods['/'])) { if ($request->isOption) {
return null; return static::$nodes[$method]['*'] ?? null;
} }
return $methods['/']; return null;
} }
HttpServer/Server.php
+8 -5
View File
@@ -92,11 +92,14 @@ class Server extends HttpService
*/ */
private function rpcListener($rpcService) private function rpcListener($rpcService)
{ {
$rpcService['events'][Constant::CONNECT] = [Service::class, 'onConnect']; if (in_array($rpcService['mode'], [SWOOLE_SOCK_UDP, SWOOLE_UDP, SWOOLE_UDP6, SWOOLE_SOCK_UDP6])) {
$rpcService['events'][Constant::DISCONNECT] = [Service::class, 'onClose']; $rpcService['events'][Constant::PACKET] = [Service::class, 'onPacket'];
$rpcService['events'][Constant::CLOSE] = [Service::class, 'onClose']; } else {
$rpcService['events'][Constant::RECEIVE] = [Service::class, 'onReceive']; $rpcService['events'][Constant::RECEIVE] = [Service::class, 'onReceive'];
$rpcService['events'][Constant::PACKET] = [Service::class, 'onPacket']; $rpcService['events'][Constant::CONNECT] = [Service::class, 'onConnect'];
$rpcService['events'][Constant::DISCONNECT] = [Service::class, 'onDisconnect'];
$rpcService['events'][Constant::CLOSE] = [Service::class, 'onClose'];
}
$this->manager->addListener($rpcService['type'], $rpcService['host'], $rpcService['port'], $rpcService['mode'], $rpcService); $this->manager->addListener($rpcService['type'], $rpcService['host'], $rpcService['port'], $rpcService['mode'], $rpcService);
} }
Rpc/Service.php
+82 -90
View File
@@ -7,15 +7,9 @@ use Exception;
use HttpServer\Http\Context; use HttpServer\Http\Context;
use HttpServer\Http\Request; use HttpServer\Http\Request;
use HttpServer\Route\Router; use HttpServer\Route\Router;
use HttpServer\Service\Http; use Server\Constant;
use HttpServer\Service\Packet;
use HttpServer\Service\Receive;
use HttpServer\Service\Websocket;
use Snowflake\Abstracts\Component;
use Snowflake\Abstracts\Config;
use Snowflake\Core\Json; use Snowflake\Core\Json;
use Snowflake\Event; use Snowflake\Event;
use Snowflake\Exception\ConfigException;
use Snowflake\Snowflake; use Snowflake\Snowflake;
use Swoole\Server; use Swoole\Server;
use function Swoole\Coroutine\defer; use function Swoole\Coroutine\defer;
@@ -25,32 +19,22 @@ use function Swoole\Coroutine\defer;
* Class Service * Class Service
* @package Rpc * @package Rpc
*/ */
class Service extends Component class Service extends \Server\Abstracts\Server
{ {
const defaultConfig = [ const RPC_CONNECT = 'RPC::CONNECT';
'open_tcp_keepalive' => true, const RPC_CLOSE = 'RPC::CLOSE';
'tcp_keepidle' => 30,
'tcp_keepinterval' => 10, private Router $router;
'tcp_keepcount' => 10,
'open_http_protocol' => false,
'open_websocket_protocol' => false,
];
const RPC_CONNECT = 'RPC::CONNECT'; /**
const RPC_CLOSE = 'RPC::CLOSE'; * @throws Exception
*/
private Router $router; public function init()
{
$this->router = Snowflake::getApp('router');
/** }
* @throws Exception
*/
public function init()
{
$this->router = Snowflake::getApp('router');
}
/** /**
@@ -59,16 +43,12 @@ class Service extends Component
* @param int $reactorId * @param int $reactorId
* @throws Exception * @throws Exception
*/ */
public function onConnect(Server $server, int $fd, int $reactorId) public function onConnect(Server $server, int $fd, int $reactorId)
{ {
defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES)); defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES));
$config = $server->setting['enable_delay_receive'] ?? null; $this->runEvent(Constant::CONNECT, null, [$server, $fd, $reactorId]);
if ($config === true) { }
$server->confirm($fd);
}
Event::trigger(Service::RPC_CONNECT, [$server, $fd, $reactorId]);
}
/** /**
@@ -77,37 +57,50 @@ class Service extends Component
* on tcp client close * on tcp client close
* @throws Exception * @throws Exception
*/ */
public function onClose(Server $server, int $fd) public function onClose(Server $server, int $fd)
{ {
defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES)); defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES));
Event::trigger(Service::RPC_CLOSE, [$server, $fd]); $this->runEvent(Constant::CLOSE, null, [$server, $fd]);
} }
/** /**
* @param Server $server * @param Server $server
* @param int $fd * @param int $fd
* @param int $reID * on tcp client close
* @param string $data * @throws Exception
* @throws Exception */
*/ public function onDisconnect(Server $server, int $fd)
public function onReceive(Server $server, int $fd, int $reID, string $data) {
{ defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES));
defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES));
try {
$client = $server->getClientInfo($fd, $reID);
$request = $this->requestSpl((int)$client['server_port'], $data); $this->runEvent(Constant::DISCONNECT, null, [$server, $fd]);
}
$result = $this->router->find_path($request)?->dispatch();
$server->send($fd, $result); /**
} catch (\Throwable $exception) { * @param Server $server
$this->addError($exception, 'rpc-service'); * @param int $fd
$server->send($fd, $exception->getMessage()); * @param int $reID
} * @param string $data
} * @throws Exception
*/
public function onReceive(Server $server, int $fd, int $reID, string $data)
{
defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES));
try {
$client = $server->getClientInfo($fd, $reID);
$request = $this->requestSpl((int)$client['server_port'], $data);
$result = $this->router->find_path($request)?->dispatch();
$server->send($fd, $result);
} catch (\Throwable $exception) {
$server->send($fd, $exception->getMessage());
}
}
/** /**
@@ -116,20 +109,19 @@ class Service extends Component
* @param array $client * @param array $client
* @throws Exception * @throws Exception
*/ */
public function onPacket(Server $server, string $data, array $client) public function onPacket(Server $server, string $data, array $client)
{ {
defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES)); defer(fn() => fire(Event::SYSTEM_RESOURCE_RELEASES));
try { try {
$request = $this->requestSpl((int)$client['server_port'], $data); $request = $this->requestSpl((int)$client['server_port'], $data);
$result = $this->router->find_path($request)?->dispatch(); $result = $this->router->find_path($request)?->dispatch();
$server->sendto($client['address'], $client['port'], $result); $server->sendto($client['address'], $client['port'], $result);
} catch (\Throwable $exception) { } catch (\Throwable $exception) {
$this->addError($exception, 'rpc-service'); $server->sendto($client['address'], $client['port'], $exception->getMessage());
$server->sendto($client['address'], $client['port'], $exception->getMessage()); }
} }
}
/** /**
@@ -138,25 +130,25 @@ class Service extends Component
* @return mixed * @return mixed
* @throws Exception * @throws Exception
*/ */
public function requestSpl(int $server_port, string $data): mixed public function requestSpl(int $server_port, string $data): mixed
{ {
$sRequest = new Request(); $sRequest = new Request();
[$cmd, $repeat, $body] = explode("\n", $data); [$cmd, $repeat, $body] = explode("\n", $data);
if (is_null($body) || is_null($cmd) || !empty($repeat)) { if (is_null($body) || is_null($cmd) || !empty($repeat)) {
throw new Exception('Protocol format error.'); throw new Exception('Protocol format error.');
} }
if (is_string($body) && is_null($data = Json::decode($body))) { if (is_string($body) && is_null($data = Json::decode($body))) {
throw new Exception('Protocol format error.'); throw new Exception('Protocol format error.');
} }
$sRequest->params->setPosts($data); $sRequest->params->setPosts($data);
$sRequest->headers->setRequestUri('rpc/p' . $server_port . '/' . ltrim($cmd, '/')); $sRequest->headers->setRequestUri('rpc/p' . $server_port . '/' . ltrim($cmd, '/'));
$sRequest->headers->setRequestMethod(Request::HTTP_CMD); $sRequest->headers->setRequestMethod(Request::HTTP_CMD);
return Context::setContext('request', $sRequest); return Context::setContext('request', $sRequest);
} }
} }