Optimized TcpServer::onReceive

Author: limingxinleo

composer.json

@@ -19,7 +19,8 @@
     },
     "require": {
         "php": ">=7.3",
-        "multiplex/socket": "^0.1"
+        "multiplex/socket": "^0.1",
+        "simple-swoole/simps": "^1.0"
     },
     "require-dev": {
         "friendsofphp/php-cs-fixer": "^2.14",

src/Protocol.php

@@ -28,6 +28,16 @@ class Protocol
      */
     public $params;
 
+    /**
+     * @var mixed
+     */
+    public $result;
+
+    /**
+     * @var array
+     */
+    public $error;
+
     public function __construct(string $class, string $method, array $params)
     {
         $this->class = $class;
@@ -49,4 +59,30 @@ public function getParams(): array
     {
         return $this->params;
     }
+
+    public function getResult()
+    {
+        return $this->result;
+    }
+
+    public function setResult($result)
+    {
+        $this->result = $result;
+        return $this;
+    }
+
+    public function setError(int $code, string $message)
+    {
+        $this->error = [
+            'code' => $code,
+            'message' => $message,
+        ];
+
+        return $this;
+    }
+
+    public function getError(): array
+    {
+        return $this->error;
+    }
 }

src/TcpServer.php

@@ -12,6 +12,7 @@
 namespace Simps\RpcMultiplex;
 
 use Multiplex\Packer;
+use Multiplex\Packet;
 use Multiplex\Serializer\StringSerializer;
 use Simps\Application;
 use Simps\Listener;
@@ -21,11 +22,19 @@
 
 class TcpServer
 {
+    /**
+     * @var Server
+     */
     protected $_server;
 
+    /**
+     * @var array
+     */
     protected $_config;
 
-    /** @var \Simps\Route */
+    /**
+     * @var Route
+     */
     protected $_route;
 
     /**
@@ -75,13 +84,13 @@ public function __construct()
 
     public function onStart(Server $server)
     {
-        Application::echoSuccess("Swoole WebSocket Server running：ws://{$this->_config['ip']}:{$this->_config['port']}");
+        Application::echoSuccess("Swoole Multiplex RPC Server running：tcp://{$this->_config['ip']}:{$this->_config['port']}");
         Listener::getInstance()->listen('start', $server);
     }
 
     public function onManagerStart(Server $server)
     {
-        Application::echoSuccess("Swoole WebSocket Server running：ws://{$this->_config['ip']}:{$this->_config['port']}");
+        Application::echoSuccess("Swoole Multiplex RPC Server running：tcp://{$this->_config['ip']}:{$this->_config['port']}");
         Listener::getInstance()->listen('managerStart', $server);
     }
 
@@ -93,14 +102,23 @@ public function onWorkerStart(Server $server, int $workerId)
 
     public function onReceive(Server $server, int $fd, int $fromId, string $data)
     {
-        Coroutine::create(function () use ($server, $fd, $fromId, $data) {
+        Coroutine::create(function () use ($server, $fd, $data) {
             $packet = $this->packer->unpack($data);
-
             $id = $packet->getId();
-            $body = $packet->getBody();
-
-            /** @var Protocol $protocol */
-            $protocol = unserialize($body);
+            try {
+                /** @var Protocol $protocol */
+                $protocol = unserialize($packet->getBody());
+
+                $class = $protocol->getClass();
+                $method = $protocol->getMethod();
+                $params = $protocol->getParams();
+                $protocol->setResult((new $class())->{$method}(...$params));
+                $result = serialize($protocol);
+            } catch (\Throwable $exception) {
+                $result = serialize($protocol->setError($exception->getCode(), $exception->getMessage()));
+            } finally {
+                $server->send($fd, $this->packer->pack(new Packet($id, $this->serializer->serialize($result))));
+            }
         });
     }
 }