Merge pull request #6 from spiral/feature/protobuf

Add protobuf support
roadrunner-php · Jun 10, 2021 · 2238be1 · 2238be1
2 parents 3985b95 + 072ea05
commit 2238be1
Show file tree

Hide file tree

Showing 10 changed files with 190 additions and 35 deletions.
diff --git a/composer.json b/composer.json
@@ -18,7 +18,8 @@
     "require-dev": {
         "vimeo/psalm": "^4.6",
         "spiral/code-style": "^1.0",
-        "rybakit/msgpack": "^0.7.1",
+        "google/protobuf": "^3.17",
+        "rybakit/msgpack": "^0.7",
         "phpunit/phpunit": "^8.0",
         "jetbrains/phpstorm-attributes": "^1.0"
     },
@@ -34,7 +35,9 @@
     },
     "suggest": {
         "ext-msgpack": "MessagePack codec support",
-        "rybakit/msgpack": "(^0.7) MessagePack codec support"
+        "ext-protobuf": "Protobuf codec support",
+        "rybakit/msgpack": "(^0.7) MessagePack codec support",
+        "google/protobuf": "(^3.0) Protobuf codec support"
     },
     "scripts": {
         "test": "phpunit --no-coverage --colors=always",

diff --git a/resources/.phpstorm.meta.php b/resources/.phpstorm.meta.php
@@ -0,0 +1,53 @@
+<?php
+
+namespace PHPSTORM_META {
+
+    //
+    // Relay
+    //
+
+    registerArgumentsSet('goridge_relay_socket_type',
+        \Spiral\Goridge\SocketRelay::SOCK_TCP,
+        \Spiral\Goridge\SocketRelay::SOCK_UNIX,
+    );
+
+    expectedArguments(\Spiral\Goridge\SocketRelay::__construct(), 2, 'goridge_relay_socket_type');
+
+    //
+    // RPC
+    //
+
+    registerArgumentsSet('goridge_rpc_options_json',
+        \JSON_BIGINT_AS_STRING,
+        \JSON_INVALID_UTF8_IGNORE,
+        \JSON_INVALID_UTF8_SUBSTITUTE,
+        \JSON_OBJECT_AS_ARRAY,
+        \JSON_THROW_ON_ERROR,
+    );
+
+    registerArgumentsSet('goridge_rpc_options_msgpack',
+        \MessagePack\UnpackOptions::BIGINT_AS_DEC,
+        \MessagePack\UnpackOptions::BIGINT_AS_GMP,
+        \MessagePack\UnpackOptions::BIGINT_AS_STR,
+    );
+
+    expectedArguments(\Spiral\Goridge\RPC\RPCInterface::call(), 2, 'goridge_rpc_options_json');
+    expectedArguments(\Spiral\Goridge\RPC\RPC::call(), 2, 'goridge_rpc_options_json');
+    expectedArguments(\Spiral\Goridge\RPC\RPCInterface::call(), 2, 'goridge_rpc_options_msgpack');
+    expectedArguments(\Spiral\Goridge\RPC\RPC::call(), 2, 'goridge_rpc_options_msgpack');
+
+    override(\Spiral\Goridge\RPC\RPCInterface::call(), map(['' => '@']));
+    override(\Spiral\Goridge\RPC\RPC::call(), map(['' => '@']));
+
+    //
+    // RPC Methods
+    //
+
+    registerArgumentsSet('goridge_rpc_methods_informer',
+        'informer.Workers',
+        'informer.List',
+    );
+
+    expectedArguments(\Spiral\Goridge\RPC\RPCInterface::call(), 0, 'goridge_rpc_methods_informer');
+    expectedArguments(\Spiral\Goridge\RPC\RPC::call(), 0, 'goridge_rpc_methods_informer');
+}
diff --git a/src/Frame.php b/src/Frame.php
@@ -51,6 +51,7 @@ final class Frame
     public const CODEC_JSON    = 0x08;
     public const CODEC_MSGPACK = 0x10;
     public const CODEC_GOB     = 0x20;
+    public const CODEC_PROTO   = 0x80;
     /**#@-*/
 
     /**

diff --git a/src/RPC/Codec/JsonCodec.php b/src/RPC/Codec/JsonCodec.php
@@ -17,9 +17,7 @@
 final class JsonCodec implements CodecInterface
 {
     /**
-     * Coded index, uniquely identified by remote server.
-     *
-     * @return int
+     * {@inheritDoc}
      */
     public function getIndex(): int
     {
@@ -43,10 +41,16 @@ public function encode($payload): string
     /**
      * {@inheritDoc}
      */
-    public function decode(string $payload)
+    public function decode(string $payload, $options = null)
     {
         try {
-            return \json_decode($payload, true, 512, \JSON_THROW_ON_ERROR);
+            $flags = \JSON_THROW_ON_ERROR;
+
+            if (\is_int($options)) {
+                $flags |= $options;
+            }
+
+            return \json_decode($payload, true, 512, $flags);
         } catch (\JsonException $e) {
             throw new CodecException(\sprintf('Json decode: %s', $e->getMessage()), (int)$e->getCode(), $e);
         }

diff --git a/src/RPC/Codec/MsgpackCodec.php b/src/RPC/Codec/MsgpackCodec.php
@@ -16,7 +16,7 @@
 
 /**
  * @psalm-type PackHandler = \Closure(mixed): string
- * @psalm-type UnpackHandler = \Closure(string): mixed
+ * @psalm-type UnpackHandler = \Closure(string, mixed|null): mixed
  */
 final class MsgpackCodec implements CodecInterface
 {
@@ -41,35 +41,33 @@ public function __construct()
     }
 
     /**
-     * Coded index, uniquely identified by remote server.
-     *
-     * @return int
+     * {@inheritDoc}
      */
     public function getIndex(): int
     {
         return Frame::CODEC_MSGPACK;
     }
 
     /**
-     * @param mixed $payload
-     * @return string
+     * {@inheritDoc}
      */
     public function encode($payload): string
     {
         return ($this->pack)($payload);
     }
 
     /**
-     * @param string $payload
-     * @return mixed
+     * {@inheritDoc}
      */
-    public function decode(string $payload)
+    public function decode(string $payload, $options = null)
     {
-        return ($this->unpack)($payload);
+        return ($this->unpack)($payload, $options);
     }
 
     /**
      * Init pack and unpack functions.
+     *
+     * @psalm-suppress MixedArgument
      */
     private function initPacker(): void
     {
@@ -79,7 +77,11 @@ private function initPacker(): void
                 return msgpack_pack($payload);
             };
 
-            $this->unpack = static function (string $payload) {
+            $this->unpack = static function (string $payload, $options = null) {
+                if ($options !== null) {
+                    return msgpack_unpack($payload, $options);
+                }
+
                 return msgpack_unpack($payload);
             };
 
@@ -92,8 +94,8 @@ private function initPacker(): void
                 return MessagePack::pack($payload);
             };
 
-            $this->unpack = static function (string $payload) {
-                return MessagePack::unpack($payload);
+            $this->unpack = static function (string $payload, $options = null) {
+                return MessagePack::unpack($payload, $options);
             };
         }
 

diff --git a/src/RPC/Codec/ProtobufCodec.php b/src/RPC/Codec/ProtobufCodec.php
@@ -0,0 +1,93 @@
+<?php
+
+/**
+ * Dead simple, high performance, drop-in bridge to Golang RPC with zero dependencies
+ *
+ * @author Wolfy-J
+ */
+
+declare(strict_types=1);
+
+namespace Spiral\Goridge\RPC\Codec;
+
+use Spiral\Goridge\Frame;
+use Google\Protobuf\Internal\Message;
+use Spiral\Goridge\RPC\CodecInterface;
+
+final class ProtobufCodec implements CodecInterface
+{
+    /**
+     * @var string
+     */
+    private const ERROR_DEPENDENCY =
+        'Could not initialize protobuf codec. ' .
+        'Please add "ext-protobuf" PECL extension or ' .
+        'install "google/protobuf" Composer dependency.';
+
+    public function __construct()
+    {
+        $this->assertAvailable();
+    }
+
+    /**
+     * @return void
+     */
+    private function assertAvailable(): void
+    {
+        if (!\class_exists(Message::class)) {
+            throw new \LogicException(self::ERROR_DEPENDENCY);
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getIndex(): int
+    {
+        return Frame::CODEC_PROTO;
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * @psalm-suppress MixedInferredReturnType
+     * @psalm-suppress MixedReturnStatement
+     */
+    public function encode($payload): string
+    {
+        if ($payload instanceof Message) {
+            return $payload->serializeToString();
+        }
+
+        return $payload;
+    }
+
+    /**
+     * @psalm-suppress UnsafeInstantiation
+     *
+     * @param class-string<Message> $class
+     * @return Message
+     */
+    protected function create(string $class): Message
+    {
+        return new $class();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function decode(string $payload, $options = null)
+    {
+        if (\is_string($options) && \is_subclass_of($options, Message::class, true)) {
+            $options = $this->create($options);
+        }
+
+        if ($options instanceof Message) {
+            $options->mergeFromString($payload);
+
+            return $options;
+        }
+
+        return $payload;
+    }
+}
diff --git a/src/RPC/Codec/RawCodec.php b/src/RPC/Codec/RawCodec.php
@@ -17,18 +17,15 @@
 final class RawCodec implements CodecInterface
 {
     /**
-     * Coded index, uniquely identified by remote server.
-     *
-     * @return int
+     * {@inheritDoc}
      */
     public function getIndex(): int
     {
         return Frame::CODEC_RAW;
     }
 
     /**
-     * @param mixed $payload
-     * @return string
+     * {@inheritDoc}
      */
     public function encode($payload): string
     {
@@ -42,10 +39,9 @@ public function encode($payload): string
     }
 
     /**
-     * @param string $payload
-     * @return mixed
+     * {@inheritDoc}
      */
-    public function decode(string $payload)
+    public function decode(string $payload, $options = null)
     {
         return $payload;
     }

diff --git a/src/RPC/CodecInterface.php b/src/RPC/CodecInterface.php
@@ -33,8 +33,9 @@ public function encode($payload): string;
 
     /**
      * @param string $payload
+     * @param mixed|null $options
      * @return mixed
      * @throws CodecException
      */
-    public function decode(string $payload);
+    public function decode(string $payload, $options = null);
 }
diff --git a/src/RPC/RPC.php b/src/RPC/RPC.php
@@ -78,7 +78,7 @@ public function withCodec(CodecInterface $codec): RPCInterface
     /**
      * {@inheritDoc}
      */
-    public function call(string $method, $payload)
+    public function call(string $method, $payload, $options = null)
     {
         $this->relay->send($this->packFrame($method, $payload));
 
@@ -95,7 +95,7 @@ public function call(string $method, $payload)
 
         self::$seq++;
 
-        return $this->decodeResponse($frame);
+        return $this->decodeResponse($frame, $options);
     }
 
     /**
@@ -112,11 +112,12 @@ public static function create(string $connection, CodecInterface $codec = null):
 
     /**
      * @param Frame $frame
+     * @param mixed|null $options
      * @return mixed
      *
      * @throws Exception\ServiceException
      */
-    private function decodeResponse(Frame $frame)
+    private function decodeResponse(Frame $frame, $options = null)
     {
         // exclude method name
         $body = \substr((string)$frame->payload, $frame->options[1]);
@@ -127,7 +128,7 @@ private function decodeResponse(Frame $frame)
             throw new ServiceException(\sprintf("Error '%s' on %s", $body, $name));
         }
 
-        return $this->codec->decode($body);
+        return $this->codec->decode($body, $options);
     }
 
     /**

diff --git a/src/RPC/RPCInterface.php b/src/RPC/RPCInterface.php
@@ -37,10 +37,11 @@ public function withCodec(CodecInterface $codec): self;
      * Invoke remove RoadRunner service method using given payload (free form).
      *
      * @param string $method
-     * @param mixed  $payload
+     * @param mixed $payload
+     * @param mixed|null $options
      * @return mixed
      * @throws GoridgeException
      * @throws RPCException
      */
-    public function call(string $method, $payload);
+    public function call(string $method, $payload, $options = null);
 }