41 Commits

Author SHA1 Message Date
as2252258 8824f711bb Revert "改名"
This reverts commit fdf58326
2022-01-20 19:04:16 +08:00
as2252258 f2d97832b5 Revert "改名"
This reverts commit fdf58326
2022-01-19 17:27:47 +08:00
as2252258 855e57a651 Revert "改名"
This reverts commit fdf58326
2022-01-19 16:39:29 +08:00
as2252258 0b3310b82e Revert "改名"
This reverts commit fdf58326
2022-01-14 11:29:16 +08:00
as2252258 e8cf7f29f0 Revert "改名"
This reverts commit fdf58326
2022-01-13 19:02:07 +08:00
as2252258 62e3b001f0 Revert "改名"
This reverts commit fdf58326
2022-01-13 19:00:01 +08:00
as2252258 f655cebdf3 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:58:40 +08:00
as2252258 ee492d97ad Revert "改名"
This reverts commit fdf58326
2022-01-13 18:57:53 +08:00
as2252258 a16d841d98 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:56:41 +08:00
as2252258 f015662fd6 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:52:42 +08:00
as2252258 021dcae59b Revert "改名"
This reverts commit fdf58326
2022-01-13 18:50:19 +08:00
as2252258 c6dc34c821 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:47:34 +08:00
as2252258 88c3958aba Revert "改名"
This reverts commit fdf58326
2022-01-13 18:42:42 +08:00
as2252258 24ebfac360 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:41:16 +08:00
as2252258 b3c200b443 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:34:23 +08:00
as2252258 3e2b50bf60 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:33:56 +08:00
as2252258 89c142930e Revert "改名"
This reverts commit fdf58326
2022-01-13 18:29:51 +08:00
as2252258 1515d9695c Revert "改名"
This reverts commit fdf58326
2022-01-13 18:28:22 +08:00
as2252258 f1f7b1081d Revert "改名"
This reverts commit fdf58326
2022-01-13 18:27:36 +08:00
as2252258 07533e29f9 Revert "改名"
This reverts commit fdf58326
2022-01-13 18:25:29 +08:00
as2252258 4d60a24fac Revert "改名"
This reverts commit fdf58326
2022-01-12 14:10:33 +08:00
as2252258 4dc6bc661a Revert "改名"
This reverts commit fdf58326
2022-01-11 17:56:16 +08:00
as2252258 0d7fd5e356 Revert "改名"
This reverts commit fdf58326
2022-01-11 14:53:41 +08:00
as2252258 da16b0685d Revert "改名"
This reverts commit fdf58326
2022-01-11 14:51:50 +08:00
as2252258 febbdea8c8 Revert "改名"
This reverts commit fdf58326
2022-01-11 14:44:02 +08:00
as2252258 a45c3d875c Revert "改名"
This reverts commit fdf58326
2022-01-11 14:42:04 +08:00
as2252258 64d01c0a80 Revert "改名"
This reverts commit fdf58326
2022-01-11 14:41:06 +08:00
as2252258 bf5fe594e4 Revert "改名"
This reverts commit fdf58326
2022-01-11 14:30:58 +08:00
as2252258 5cfe3c6d6d Revert "改名"
This reverts commit fdf58326
2022-01-11 14:30:05 +08:00
as2252258 773e3c0f57 Revert "改名"
This reverts commit fdf58326
2022-01-11 14:26:57 +08:00
as2252258 b1c91343ef Revert "改名"
This reverts commit fdf58326
2022-01-11 14:25:24 +08:00
as2252258 3f821ca9d0 Revert "改名"
This reverts commit fdf58326
2022-01-11 14:23:48 +08:00
as2252258 d296b3107e Revert "改名"
This reverts commit fdf58326
2022-01-11 14:15:51 +08:00
as2252258 f606c58204 Revert "改名"
This reverts commit fdf58326
2022-01-11 14:13:51 +08:00
as2252258 6c1d986814 Revert "改名"
This reverts commit fdf58326
2022-01-10 11:39:56 +08:00
as2252258 da056b09c7 1 2022-01-09 16:07:57 +08:00
as2252258 4eefe54be0 e 2022-01-09 14:00:32 +08:00
as2252258 b0ed7650c5 Revert "改名"
This reverts commit fdf58326
2022-01-08 18:49:07 +08:00
as2252258 3d89cbd4c6 Revert "改名"
This reverts commit fdf58326
2022-01-05 15:22:29 +08:00
as2252258 0f2d19f3c7 Revert "改名"
This reverts commit fdf58326
2022-01-05 15:21:48 +08:00
as2252258 9001dd041e Revert "改名"
This reverts commit fdf58326
2022-01-04 15:56:37 +08:00
19 changed files with 1519 additions and 1484 deletions
@@ -1,13 +1,13 @@
<?php
namespace Kiri\Rpc\Note;
namespace Kiri\Rpc\Annotation;
use Kiri\Abstracts\Config;
use Kiri\Core\Network;
use Kiri\Exception\ConfigException;
use Kiri\Kiri;
use Kiri;
use Kiri\Rpc\RpcManager;
use Note\Attribute;
use Kiri\Annotation\Attribute;
use ReflectionException;
#[\Attribute(\Attribute::TARGET_CLASS)] class JsonRpc extends Attribute
@@ -23,10 +23,11 @@ use ReflectionException;
* @param array $tags
* @param array $meta
* @param array $checkOptions
* @param string $checkUrl
*/
public function __construct(public string $service, public string $driver, public array $tags = [], public array $meta = [], public array $checkOptions = [])
public function __construct(public string $service, public string $driver, public array $tags = [], public array $meta = [], public array $checkOptions = [], public string $checkUrl = '')
{
$this->uniqueId = preg_replace('/(\w{11})(\w{4})(\w{3})(\w{8})(\w{6})/', '$1-$2-$3-$4-$5', md5(__DIR__ . '.' . md5(Network::local())));
$this->uniqueId = preg_replace('/(\w{11})(\w{4})(\w{3})(\w{8})(\w{6})/', '$1-$2-$3-$4-$5', md5(__DIR__ . 'Annotation' . md5(Network::local())));
}
@@ -49,6 +50,9 @@ use ReflectionException;
protected function create(): array
{
$rpcPort = Config::get('rpc.port');
if (empty($this->checkUrl)) {
$this->checkUrl = Network::local() . ":" . Config::get('rpc.port');
}
$defaultConfig = [
"ID" => "rpc.json.{$this->service}." . $this->uniqueId,
"Name" => $this->service,
@@ -68,7 +72,7 @@ use ReflectionException;
"Name" => "service " . $this->service . ' health check',
"Notes" => "Script based health check",
"ServiceID" => $this->service,
"TCP" => Network::local() . ":" . Config::get('rpc.port'),
"TCP" => $this->checkUrl,
"Interval" => "5s",
"Timeout" => "1s",
"DeregisterCriticalServiceAfter" => "30s"
+1 -1
View File
@@ -6,7 +6,7 @@ use Exception;
use Kiri\Abstracts\Component;
use Kiri\Context;
use Kiri\Exception\ConfigException;
use Kiri\Kiri;
use Kiri;
use Kiri\Pool\Alias;
use Kiri\Pool\Pool;
use Swoole\Client;
@@ -4,6 +4,7 @@ namespace Kiri\Rpc;
use JetBrains\PhpStorm\Pure;
use Throwable;
class InvalidRpcParamsException extends \Exception
@@ -15,7 +16,7 @@ class InvalidRpcParamsException extends \Exception
* @param int $code
* @param Throwable|null $previous
*/
public function __construct($message = "", $code = 0, Throwable $previous = null)
#[Pure] public function __construct(string $message = "", int $code = 0, Throwable $previous = null)
{
parent::__construct($message, -32602, $previous);
}
@@ -4,10 +4,10 @@ namespace Kiri\Rpc;
use Exception;
use Http\Message\ServerRequest;
use Http\Message\Stream;
use Kiri\Message\ServerRequest;
use Kiri\Message\Stream;
use Kiri\Core\Number;
use Kiri\Kiri;
use Kiri;
use Kiri\Pool\Pool;
use Psr\Http\Client\ClientExceptionInterface;
use Psr\Http\Message\ResponseInterface;
@@ -54,7 +54,6 @@ abstract class JsonRpcConsumers implements OnRpcConsumerInterface
/**
* @param array $data
* @return ServerRequestInterface
* @throws \ReflectionException
*/
private function requestBody(array $data): ServerRequestInterface
{
@@ -2,10 +2,10 @@
namespace Kiri\Rpc;
use Note\Inject;
use Kiri\Annotation\Inject;
use Exception;
use Http\Message\Response;
use Http\Message\Stream;
use Kiri\Message\Response;
use Kiri\Message\Stream;
use Kiri\Abstracts\Config;
use Kiri\Exception\ConfigException;
use Psr\Http\Client\ClientInterface;
@@ -2,8 +2,8 @@
namespace Kiri\Rpc;
use Http\Message\Response;
use Http\Message\Stream;
use Kiri\Message\Response;
use Kiri\Message\Stream;
use Psr\Http\Client\ClientInterface;
use Psr\Http\Message\RequestInterface;
use Psr\Http\Message\ResponseInterface;
View File
+39 -34
View File
@@ -2,30 +2,29 @@
namespace Kiri\Rpc;
use Http\Constrict\RequestInterface;
use Http\Handler\Handler;
use Http\Handler\Router;
use Http\Message\ServerRequest;
use Kiri;
use Kiri\Abstracts\Component;
use Kiri\Abstracts\Config;
use Kiri\Annotation\Annotation;
use Kiri\Annotation\Inject;
use Kiri\Consul\Agent;
use Kiri\Context;
use Kiri\Events\EventProvider;
use Kiri\Exception\ConfigException;
use Kiri\Kiri;
use Note\Inject;
use Note\Note;
use Kiri\Message\Constrict\RequestInterface;
use Kiri\Message\Handler\Handler;
use Kiri\Message\Handler\Router;
use Kiri\Message\ServerRequest;
use Kiri\Server\Contract\OnCloseInterface;
use Kiri\Server\Contract\OnConnectInterface;
use Kiri\Server\Contract\OnReceiveInterface;
use Kiri\Server\Events\OnBeforeShutdown;
use Kiri\Server\Events\OnServerBeforeStart;
use Kiri\Server\Events\OnTaskerStart;
use Kiri\Server\Events\OnWorkerExit;
use Kiri\Server\Events\OnWorkerStart;
use Psr\Container\ContainerExceptionInterface;
use Psr\Container\NotFoundExceptionInterface;
use Psr\Http\Message\ServerRequestInterface;
use ReflectionException;
use Server\Contract\OnCloseInterface;
use Server\Contract\OnConnectInterface;
use Server\Contract\OnReceiveInterface;
use Server\Events\OnBeforeShutdown;
use Server\Events\OnServerBeforeStart;
use Server\Events\OnTaskerStart;
use Server\Events\OnWorkerStart;
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Swoole\Server;
@@ -43,25 +42,28 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
public Router $router;
#[Inject(Note::class)]
public Note $annotation;
#[Inject(Annotation::class)]
public Annotation $annotation;
private RpcManager $manager;
private int $timerId;
/**
*
* @throws \Exception
*/
public function init(): void
{
$this->eventProvider->on(OnBeforeShutdown::class, [$this, 'onBeforeShutdown']);
$this->getEventProvider()->on(OnBeforeShutdown::class, [$this, 'onBeforeShutdown']);
scan_directory(APP_PATH . 'rpc', 'Rpc');
scan_directory(APP_PATH . 'rpc', 'app\Rpc');
$this->eventProvider->on(OnWorkerStart::class, [$this, 'consulWatches']);
$this->eventProvider->on(OnServerBeforeStart::class, [$this, 'register']);
$this->getEventProvider()->on(OnWorkerStart::class, [$this, 'consulWatches']);
$this->getEventProvider()->on(OnWorkerExit::class, [$this, 'onWorkerExit']);
$this->getEventProvider()->on(OnServerBeforeStart::class, [$this, 'register']);
$this->manager = Kiri::getDi()->get(RpcManager::class);
}
@@ -72,12 +74,11 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
* @return void
* @throws ContainerExceptionInterface
* @throws NotFoundExceptionInterface
* @throws ReflectionException
*/
public function onBeforeShutdown(OnBeforeShutdown $beforeShutdown)
{
$doneList = $this->manager->doneList();
$agent = $this->container->get(Agent::class);
$agent = $this->getContainer()->get(Agent::class);
foreach ($doneList as $value) {
$agent->service->deregister($value['config']['ID']);
$agent->checks->deregister($value['config']['Check']['CheckId']);
@@ -95,19 +96,24 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
return;
}
$async_time = (int)Config::get('consul.async_time', 1000);
Timer::tick($async_time, static function ($timeId) {
if (env('state', 'start') == 'exit') {
Timer::clear($timeId);
return;
}
$this->timerId = Timer::tick($async_time, static function () {
Kiri::getDi()->get(RpcManager::class)->tick();
});
}
/**
* @param OnWorkerExit $exit
* @return void
*/
public function onWorkerExit(OnWorkerExit $exit)
{
Timer::clear($this->timerId);
}
/**
* @param OnServerBeforeStart $server
* @throws ReflectionException
*/
public function register(OnServerBeforeStart $server)
{
@@ -201,7 +207,7 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
private function dispatch($data): array
{
try {
[$handler, $params] = $this->container->get(RpcManager::class)->get($data['service'], $data['method']);
[$handler, $params] = $this->getContainer()->get(RpcManager::class)->get($data['service'], $data['method']);
if (is_null($handler)) {
throw new \Exception('Method not found', -32601);
} else {
@@ -266,10 +272,9 @@ class RpcJsonp extends Component implements OnConnectInterface, OnReceiveInterfa
/**
* @param Server $server
* @param int $fd
*/
public function onClose(Server $server, int $fd): void
public function onClose(int $fd): void
{
// TODO: Implement onClose() method.
}
+35 -8
View File
@@ -2,11 +2,12 @@
namespace Kiri\Rpc;
use Http\Handler\Handler;
use Exception;
use Kiri;
use Kiri\Abstracts\Component;
use Kiri\Consul\Agent;
use Kiri\Consul\Health;
use Kiri\Kiri;
use Kiri\Message\Handler\Handler;
use ReflectionException;
class RpcManager extends Component
@@ -22,11 +23,12 @@ class RpcManager extends Component
/**
* @param $serviceName
* @return void
* @throws ReflectionException
* @throws \Exception
* @throws Exception
*/
public function async($serviceName): void
{
$this->reRegister($serviceName);
$lists = Kiri::getDi()->get(Health::class)->setQuery('passing=true')->service($serviceName);
if ($lists->getStatusCode() != 200) {
return;
@@ -42,20 +44,45 @@ class RpcManager extends Component
/**
* @throws ReflectionException
* @param string $serviceName
* @return void
* @throws Exception
*/
public function reRegister(string $serviceName)
{
$config = $this->_rpc[$serviceName] ?? [];
if (empty($config)) {
return;
}
$service = Kiri::getDi()->get(Agent::class);
$info = $service->service->service_health($config['config']['ID']);
if ($info->getStatusCode() == 200) {
return;
}
$service->service->register($config['config']);
}
/**
* @throws Exception
*/
public function tick(): void
{
try {
foreach ($this->_rpc as $name => $list) {
$this->async($name);
}
} catch (\Throwable $throwable) {
$this->logger()->error(error_trigger_format($throwable));
}
}
/**
* @param $serviceName
* @return array
* @throws \Exception
* @throws Exception
*/
public function getServices($serviceName): array
{
@@ -112,15 +139,15 @@ class RpcManager extends Component
/**
* @throws ReflectionException
*/
public function register()
{
$agent = Kiri::getDi()->get(Agent::class);
foreach ($this->_rpc as $list) {
$agent->service->deregister($list['config']['ID']);
$data = $agent->service->register($list['config']);
if ($data->getStatusCode() != 200) {
exit($data->getBody());
return;
}
}
}
+2 -3
View File
@@ -11,12 +11,11 @@
"require": {
"php": ">=8.0",
"ext-json": "*",
"psr/http-client": "^1.0",
"psr/http-message": "^1.0"
"psr/http-client": "^1.0"
},
"autoload": {
"psr-4": {
"Kiri\\Rpc\\": "src/"
"Kiri\\Rpc\\": "./"
}
},
"require-dev": {
+2 -2
View File
@@ -3,7 +3,7 @@
use Kiri\Rpc\RpcJsonp;
use Kiri\Rpc\TestRpcService;
use Server\Constant;
use Kiri\Server\Constant;
return [
'rpc' => [
@@ -61,7 +61,7 @@ return [
"node" => "t2.320",
"checkId" => "service:redis1",
"name" => "Redis health check",
"notes" => "Script based health check",
"Annotations" => "Script based health check",
"status" => "passing",
"serviceID" => "redis1",
"definition" => [