From 3d690d6d15b76f313924ce7b7f6b695e6c5c5c9c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=90=91=E6=9E=97?= <as2252258@163.com>
Date: Tue, 26 Oct 2021 18:58:09 +0800
Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 composer.json                     |   1 -
 src/Annotation/JsonRpc.php        |  19 ++++
 src/Consumers.php                 |  53 +++++++++
 src/InvalidRpcParamsException.php |  23 ++++
 src/OnJsonRpcInterface.php        |  12 ++
 src/OnRpcConsumerInterface.php    |   8 ++
 src/Protocol.php                  |  37 ------
 src/Registry.php                  |  68 -----------
 src/RpcJsonp.php                  | 183 ++++++++++++++++++++++++++++++
 src/RpcManager.php                |  32 ------
 src/RpcProvider.php               | 109 ------------------
 src/TestRpcService.php            |  33 ++++++
 src/config.php                    |  32 ++++--
 13 files changed, 352 insertions(+), 258 deletions(-)
 create mode 100644 src/Annotation/JsonRpc.php
 create mode 100644 src/Consumers.php
 create mode 100644 src/InvalidRpcParamsException.php
 create mode 100644 src/OnJsonRpcInterface.php
 create mode 100644 src/OnRpcConsumerInterface.php
 delete mode 100644 src/Protocol.php
 delete mode 100644 src/Registry.php
 create mode 100644 src/RpcJsonp.php
 delete mode 100644 src/RpcManager.php
 delete mode 100644 src/RpcProvider.php
 create mode 100644 src/TestRpcService.php

diff --git a/composer.json b/composer.json
index 2612c24..c8558c9 100644
--- a/composer.json
+++ b/composer.json
@@ -19,6 +19,5 @@
      }
   },
   "require-dev": {
-    "kwn/php-rdkafka-stubs": "^2.0"
   }
 }
diff --git a/src/Annotation/JsonRpc.php b/src/Annotation/JsonRpc.php
new file mode 100644
index 0000000..1c15003
--- /dev/null
+++ b/src/Annotation/JsonRpc.php
@@ -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')
+	{
+
+	}
+
+
+}
diff --git a/src/Consumers.php b/src/Consumers.php
new file mode 100644
index 0000000..232d08e
--- /dev/null
+++ b/src/Consumers.php
@@ -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()
+	{
+
+
+
+	}
+
+
+}
diff --git a/src/InvalidRpcParamsException.php b/src/InvalidRpcParamsException.php
new file mode 100644
index 0000000..4635e9a
--- /dev/null
+++ b/src/InvalidRpcParamsException.php
@@ -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);
+	}
+
+}
diff --git a/src/OnJsonRpcInterface.php b/src/OnJsonRpcInterface.php
new file mode 100644
index 0000000..f7a8650
--- /dev/null
+++ b/src/OnJsonRpcInterface.php
@@ -0,0 +1,12 @@
+<?php
+
+namespace Kiri\Rpc;
+
+interface OnJsonRpcInterface
+{
+
+
+//	public function execute();
+
+
+}
diff --git a/src/OnRpcConsumerInterface.php b/src/OnRpcConsumerInterface.php
new file mode 100644
index 0000000..1141b1e
--- /dev/null
+++ b/src/OnRpcConsumerInterface.php
@@ -0,0 +1,8 @@
+<?php
+
+namespace Kiri\Rpc;
+
+interface OnRpcConsumerInterface
+{
+
+}
diff --git a/src/Protocol.php b/src/Protocol.php
deleted file mode 100644
index 5bc6601..0000000
--- a/src/Protocol.php
+++ /dev/null
@@ -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)]);
-	}
-
-}
diff --git a/src/Registry.php b/src/Registry.php
deleted file mode 100644
index dacb16c..0000000
--- a/src/Registry.php
+++ /dev/null
@@ -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);
-	}
-
-
-}
diff --git a/src/RpcJsonp.php b/src/RpcJsonp.php
new file mode 100644
index 0000000..70c15a3
--- /dev/null
+++ b/src/RpcJsonp.php
@@ -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.
+	}
+}
diff --git a/src/RpcManager.php b/src/RpcManager.php
deleted file mode 100644
index 6a83e53..0000000
--- a/src/RpcManager.php
+++ /dev/null
@@ -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;
-	}
-
-}
diff --git a/src/RpcProvider.php b/src/RpcProvider.php
deleted file mode 100644
index 4dbca56..0000000
--- a/src/RpcProvider.php
+++ /dev/null
@@ -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.
-	}
-}
diff --git a/src/TestRpcService.php b/src/TestRpcService.php
new file mode 100644
index 0000000..0e0627f
--- /dev/null
+++ b/src/TestRpcService.php
@@ -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;
+	}
+}
diff --git a/src/config.php b/src/config.php
index 7ae72c2..9f33b3e 100644
--- a/src/config.php
+++ b/src/config.php
@@ -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' => ''
 			]
-
-
 		]
-
 	]
 ];