as2252258/kiri-rpc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

改名

Browse Source
This commit is contained in:
向林
2021-10-26 18:58:09 +08:00
parent b68403f581
commit 3d690d6d15
13 changed files with 352 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files
composer.json
-1
View File
@@ -19,6 +19,5 @@
}
},
"require-dev": {
"kwn/php-rdkafka-stubs": "^2.0"
}
}
src/Annotation/JsonRpc.php
+19
View File
@@ -0,0 +1,19 @@
<?php
namespace Kiri\Rpc\Annotation;
#[\Attribute(\Attribute::TARGET_CLASS)] class JsonRpc
{
/**
* @param string $method
* @param string $version
*/
public function __construct(public string $method, public string $version = '2.0')
{
}
}
src/Consumers.php
+53
View File
@@ -0,0 +1,53 @@
<?php
namespace Kiri\Rpc;
use Kiri\Pool\Pool;
/**
*
*/
class Consumers implements OnRpcConsumerInterface
{
/**
* @var Pool
*/
public Pool $pool;
/**
* @param string $method
* @param mixed $data
* @param string $version
*/
public function notify(string $method, mixed $data, string $version = '2.0'): void
{
}
/**
* @param string $method
* @param mixed $data
* @param string $version
* @param string $id
* @return mixed
*/
public function get(string $method, mixed $data, string $version = '2.0', string $id = ''): mixed
{
}
private function get_consul()
{
}
}
src/InvalidRpcParamsException.php
+23
View File
@@ -0,0 +1,23 @@
<?php
namespace Kiri\Rpc;
use Throwable;
class InvalidRpcParamsException extends \Exception
{
/**
* @param string $message
* @param int $code
* @param Throwable|null $previous
*/
public function __construct($message = "", $code = 0, Throwable $previous = null)
{
parent::__construct($message, -32602, $previous);
}
}
src/OnJsonRpcInterface.php
+12
View File
@@ -0,0 +1,12 @@
<?php
namespace Kiri\Rpc;
interface OnJsonRpcInterface
{
// public function execute();
}
src/OnRpcConsumerInterface.php
+8
View File
@@ -0,0 +1,8 @@
<?php
namespace Kiri\Rpc;
interface OnRpcConsumerInterface
{
}
src/Protocol.php
-37
View File
@@ -1,37 +0,0 @@
<?php
namespace Kiri\Rpc;
class Protocol
{
const SPLIT_STRING = "\r\r\n\n";
/**
* @param string $data
* @return array|null
*/
public static function parse(string $data): ?array
{
if (!str_contains($data, Protocol::SPLIT_STRING)) {
return null;
}
[$cmd, $requestBody] = explode(Protocol::SPLIT_STRING, $data);
return [$cmd, json_decode($requestBody, true)];
}
/**
* @param string $cmd
* @param array $data
* @return string
*/
public static function create(string $cmd, array $data): string
{
return implode("\r\r\n\n", [$cmd, json_encode($data, JSON_UNESCAPED_UNICODE)]);
}
}
src/Registry.php
-68
View File
@@ -1,68 +0,0 @@
<?php
namespace Kiri\Rpc;
use Exception;
use Kiri\Exception\NotFindClassException;
use ReflectionException;
class Registry
{
// KV
const URI_PUT = 'kv/put';
const URI_RANGE = 'kv/range';
const URI_DELETE_RANGE = 'kv/deleterange';
const URI_TXN = 'kv/txn';
const URI_COMPACTION = 'kv/compaction';
// Lease
const URI_GRANT = 'lease/grant';
const URI_REVOKE = 'kv/lease/revoke';
const URI_KEEPALIVE = 'lease/keepalive';
const URI_TIMETOLIVE = 'kv/lease/timetolive';
// Role
const URI_AUTH_ROLE_ADD = 'auth/role/add';
const URI_AUTH_ROLE_GET = 'auth/role/get';
const URI_AUTH_ROLE_DELETE = 'auth/role/delete';
const URI_AUTH_ROLE_LIST = 'auth/role/list';
// Authenticate
const URI_AUTH_ENABLE = 'auth/enable';
const URI_AUTH_DISABLE = 'auth/disable';
const URI_AUTH_AUTHENTICATE = 'auth/authenticate';
// User
const URI_AUTH_USER_ADD = 'auth/user/add';
const URI_AUTH_USER_GET = 'auth/user/get';
const URI_AUTH_USER_DELETE = 'auth/user/delete';
const URI_AUTH_USER_CHANGE_PASSWORD = 'auth/user/changepw';
const URI_AUTH_USER_LIST = 'auth/user/list';
const URI_AUTH_ROLE_GRANT = 'auth/role/grant';
const URI_AUTH_ROLE_REVOKE = 'auth/role/revoke';
const URI_AUTH_USER_GRANT = 'auth/user/grant';
const URI_AUTH_USER_REVOKE = 'auth/user/revoke';
const PERMISSION_READ = 0;
const PERMISSION_WRITE = 1;
const PERMISSION_READWRITE = 2;
const DEFAULT_HTTP_TIMEOUT = 30;
/**
* @param $name
* @return array
* @throws NotFindClassException
* @throws ReflectionException
* @throws Exception
*/
public function getService($name): array
{
return di(Client::class)->get($name);
}
}
src/RpcJsonp.php
+183
View File
@@ -0,0 +1,183 @@
<?php
namespace Kiri\Rpc;
use Annotation\Inject;
use Http\Constrict\ResponseInterface;
use Http\Handler\Abstracts\HandlerManager;
use Http\Handler\Dispatcher;
use Http\Handler\Router;
use Http\Message\ServerRequest;
use Server\SInterface\OnCloseInterface;
use Server\SInterface\OnConnectInterface;
use Server\SInterface\OnReceiveInterface;
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Swoole\Server;
/**
*
*/
class RpcJsonp implements OnConnectInterface, OnReceiveInterface, OnCloseInterface
{
#[Inject(Router::class)]
public Router $router;
/**
* @param Server $server
* @param int $fd
*/
public function onConnect(Server $server, int $fd): void
{
// TODO: Implement onConnect() method.
}
/**
* @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): void
{
$data = json_decode($data, true);
if (is_null($data)) {
$this->failure(-32700, 'Parse error语法解析错误');
} else if (!isset($data['jsonrpc']) || !isset($data['method']) || $data['jsonrpc'] != '2.0') {
$this->failure(-32600, 'Invalid Request无效请求');
} else {
$this->batchDispatch($server, $fd, $data);
}
}
/**
* @param Server $server
* @param int $fd
* @param array $data
* @return void
*/
private function batchDispatch(Server $server, int $fd, array $data): void
{
if (isset($data['jsonrpc'])) {
$dispatch = $this->dispatch($data);
if (!isset($data['id'])) {
return;
}
$result = json_encode($dispatch, JSON_UNESCAPED_UNICODE);
} else {
$channel = new Channel($total = count($data));
foreach ($data as $datum) {
$this->_execute($channel, $datum);
}
$result = [];
for ($i = 0; $i < $total; $i++) {
$params = $channel->pop();
if (empty($params)) {
continue;
}
$result[] = $params;
}
}
$server->send($fd, json_encode($result, JSON_UNESCAPED_UNICODE));
}
/**
* @param $channel
* @param $datum
*/
private function _execute($channel, $datum)
{
Coroutine::create(function () use ($channel, $datum) {
if (empty($datum) || !isset($datum['jsonrpc'])) {
$channel->push($this->failure(-32700, 'Parse error语法解析错误'));
} else if (!isset($datum['method'])) {
$channel->push($this->failure(-32700, 'Parse error语法解析错误'));
} else {
$dispatch = $this->dispatch($datum);
if (!isset($dispatch['id'])) {
$dispatch = null;
}
$channel->push($dispatch);
}
});
}
/**
* @param $data
* @return array
*/
private function dispatch($data): array
{
try {
$handler = HandlerManager::get($data['method'], 'json-rpc');
if (is_integer($handler)) {
throw new \Exception('Invalid Request无效请求', -32600);
} else if (is_null($handler)) {
throw new \Exception('Method not found', -32601);
} else {
return $this->handler($handler, $data);
}
} catch (\Throwable $throwable) {
$code = $throwable->getCode() == 0 ? -32603 : $throwable->getCode();
return $this->failure($code, jTraceEx($throwable), [], $data['id'] ?? null);
}
}
/**
* @param $handler
* @param $data
* @return array
* @throws \Exception
*/
private function handler($handler, $data): array
{
$dispatcher = (new Dispatcher($handler, $handler->_middlewares))->handle(new ServerRequest());
if ($dispatcher instanceof ResponseInterface) {
$dispatcher = json_decode($dispatcher->getBody()->getContents(), true);
}
return ['jsonrpc' => '2.0', 'result' => $dispatcher, 'id' => $data['id'] ?? null];
}
/**
* @param $code
* @param $message
* @param array $data
* @param null $id
* @return array
*/
protected function failure($code, $message, array $data = [], $id = null): array
{
$error = [
'jsonrpc' => '2.0',
'error' => [
'code' => $code,
'message' => $message,
'data' => $data
]
];
if (!is_null($id)) {
$error['id'] = $id;
}
return $error;
}
/**
* @param Server $server
* @param int $fd
*/
public function onClose(Server $server, int $fd): void
{
// TODO: Implement onClose() method.
}
}
src/RpcManager.php
-32
View File
@@ -1,32 +0,0 @@
<?php
namespace Kiri\Rpc;
class RpcManager
{
private static array $_handler = [];
/**
* @param $cmd
* @param array $handler
* @param string $protocol
*/
public static function addCmdHandler($cmd, array $handler, string $protocol)
{
static::$_handler[$cmd] = [$handler, $protocol];
}
/**
* @param $cmd
* @return array|null
*/
public static function getHandler($cmd): ?array
{
return static::$_handler[$cmd] ?? null;
}
}
src/RpcProvider.php
-109
View File
@@ -1,109 +0,0 @@
<?php
namespace Kiri\Rpc;
use Kiri\Exception\ConfigException;
use ReflectionException;
use Server\Abstracts\Tcp;
use Server\Constant;
use Server\ServerManager;
use Server\SInterface\OnClose;
use Server\SInterface\OnConnect;
use Server\SInterface\OnReceive;
use Server\SInterface\OnRequest;
use Swoole\Http\Request;
use Swoole\Http\Response;
use Swoole\Server;
/**
*
*/
class RpcProvider extends Tcp implements OnClose, OnConnect, OnReceive, OnRequest
{
/**
* @param ServerManager $manager
* @param array $config
* @throws ConfigException
* @throws ReflectionException
*/
public static function addRpcListener(ServerManager $manager, array $config)
{
$config['settings']['enable_delay_receive'] = true;
$config['settings']['enable_unsafe_event'] = true;
$config['events'][Constant::RECEIVE] = [RpcProvider::class, 'onReceive'];
$implements = class_implements(RpcProvider::class);
if (in_array(OnConnect::class, $implements)) {
$config['events'][Constant::CONNECT] = [RpcProvider::class, 'onConnect'];
}
if (in_array(OnClose::class, $implements)) {
$config['events'][Constant::DISCONNECT] = [RpcProvider::class, 'onDisconnect'];
$config['events'][Constant::CLOSE] = [RpcProvider::class, 'onClose'];
}
$manager->addListener(
$config['type'], $config['host'], $config['port'], $config['mode'], $config
);
}
/**
* @param Server $server
* @param int $fd
*/
public function onClose(Server $server, int $fd): void
{
// TODO: Implement onClose() method.
}
/**
* @param Server $server
* @param int $fd
*/
public function onDisconnect(Server $server, int $fd): void
{
// TODO: Implement onDisconnect() method.
}
/**
* @param Server $server
* @param int $fd
*/
public function onConnect(Server $server, int $fd): void
{
$server->confirm($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): void
{
try {
[$cmd, [$body, $protocol]] = Protocol::parse($data);
} catch(\Throwable $throwable){
}
}
/**
* @param Request $request
* @param Response $response
*/
public function onRequest(Request $request, Response $response): void
{
// TODO: Implement onRequest() method.
}
}
src/TestRpcService.php
+33
View File
@@ -0,0 +1,33 @@
<?php
namespace Kiri\Rpc;
use Http\Constrict\RequestInterface;
use Kiri\Rpc\Annotation\JsonRpc;
/**
*
*/
#[JsonRpc(method: 'test.service', version: '2.0')]
class TestRpcService implements OnJsonRpcInterface
{
/**
* @var RequestInterface
*/
public RequestInterface $request;
/**
* @param int $i
* @param int $b
* @return int
*/
public function execute(int $i, int $b): int
{
return $i + $b;
}
}
src/config.php
+21 -11
View File
@@ -1,23 +1,33 @@
<?php
use Kiri\Rpc\RpcJsonp;
use Kiri\Rpc\TestRpcService;
use Server\Constant;
return [
'rpc' => [
'name' => 'json-rpc',
'type' => Constant::SERVER_TYPE_BASE,
'mode' => SWOOLE_SOCK_TCP,
'host' => '0.0.0.0',
'port' => 9526,
'settings' => [
],
'events' => [
Constant::RECEIVE => [RpcJsonp::class, 'onReceive']
],
'consumers' => [
'userService' => [
'type' => 'json',
'class' => ''
'class' => TestRpcService::class,
'name' => 'test-rpc',
'package' => 'test',
'register' => [
'host' => '',
'port' => ''
]
]
]
];