feat: Add multiplexed RPC stream transport message handling

src/ClientFactory.php

@@ -20,7 +20,7 @@ class ClientFactory
     /**
      * Create a client.
      * @param (Closure(): Metadata)|Metadata|string|mixed $service
-     * @param Metadata|string|null $metadata
+     * @param null|Metadata|string $metadata
      * @throws InvalidArgumentException
      * @throws Exception
      */

src/Consul/Response.php

@@ -23,7 +23,7 @@ class Response
     private $response;
 
     /**
-     * @var array|null
+     * @var null|array
      */
     private $decoded;
 
@@ -38,7 +38,7 @@ public function __call($name, $arguments)
     }
 
     /**
-     * @param mixed|null $default
+     * @param null|mixed $default
      * @return mixed
      * @throws ServerException
      */
@@ -60,7 +60,7 @@ public function json(?string $key = null, $default = null)
     }
 
     /**
-     * @return bool|object|null
+     * @return null|bool|object
      */
     public function object()
     {

src/Contract/RegistryInterface.php

@@ -18,7 +18,7 @@ interface RegistryInterface
     public function setLoadBalancer(?LoadBalancerInterface $loadBalancer);
 
     /**
-     * @return LoadBalancerInterface|null
+     * @return null|LoadBalancerInterface
      */
     public function getLoadBalancer();
 

src/Functions.php

@@ -21,7 +21,7 @@
  * @template T
  *
  * @param callable(int):T $callback
- * @param (callable(Throwable):mixed)|null $when
+ * @param null|(callable(Throwable):mixed) $when
  * @return T
  * @throws Throwable
  */
@@ -72,7 +72,7 @@ function throw_if($condition, $exception, ...$parameters)
  * @template TValue
  *
  * @param TValue $value
- * @param (callable(TValue):mixed)|null $callback
+ * @param null|(callable(TValue):mixed) $callback
  * @return TValue
  */
 function tap($value, ?callable $callback = null)
@@ -105,7 +105,7 @@ public function __call($method, $parameters)
  * @template TReturn
  *
  * @param TValue $value
- * @param callable(TValue):TReturn|null $callback
+ * @param null|callable(TValue):TReturn $callback
  * @return ($callback is null ? TValue : TReturn)
  */
 function with($value, ?callable $callback = null)

src/MetadataManager.php

@@ -24,7 +24,7 @@ public static function register(string $name, Metadata $metadata)
     }
 
     /**
-     * @return Metadata|null
+     * @return null|Metadata
      */
     public static function get(string $name)
     {

src/Support/Arr.php

@@ -21,7 +21,7 @@ class Arr
      * Get an item from an array using "dot" notation.
      *
      * @param array|ArrayAccess $array
-     * @param int|string|null $key
+     * @param null|int|string $key
      * @param mixed $default
      * @return mixed
      */
@@ -54,7 +54,7 @@ public static function get($array, $key = null, $default = null)
      * Check if an item or items exist in an array using "dot" notation.
      *
      * @param array|ArrayAccess $array
-     * @param array|string|null $keys
+     * @param null|array|string $keys
      * @return bool
      */
     public static function has($array, $keys)

src/Support/UserAgent.php

@@ -17,7 +17,7 @@
 class UserAgent
 {
     /**
-     * @var string|null
+     * @var null|string
      */
     protected static $value;
 

src/Transporter/AbstractTransporter.php

@@ -18,7 +18,7 @@
 abstract class AbstractTransporter implements TransporterInterface
 {
     /**
-     * @var LoadBalancerInterface|null
+     * @var null|LoadBalancerInterface
      */
     protected $loadBalancer;
 

src/Transporter/GrpcTransporter.php

@@ -35,7 +35,7 @@ class GrpcTransporter extends AbstractTransporter
     protected array $options = [];
 
     /**
-     * @var object|string|null
+     * @var null|object|string
      */
     protected $credentials;
 

src/Transporter/MultiplexRpcTransporter.php

@@ -0,0 +1,85 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of friendsofhyperf/jet.
+ *
+ * @link     https://github.com/friendsofhyperf/jet
+ * @document https://github.com/friendsofhyperf/jet/blob/main/README.md
+ * @contact  huangdijia@gmail.com
+ */
+
+namespace FriendsOfHyperf\Jet\Transporter;
+
+use Exception;
+use FriendsOfHyperf\Jet\Exception\ConnectionException;
+use FriendsOfHyperf\Jet\Exception\RecvFailedException;
+use RuntimeException;
+
+class MultiplexRpcTransporter extends StreamSocketTransporter
+{
+    public const PING = 'ping';
+
+    public const PONG = 'pong';
+
+    public function receive()
+    {
+        stream_set_blocking($this->client, false);
+
+        while (true) {
+            $header = $this->readBytes(4);
+
+            $unpacked = unpack('Nlength', $header);
+            $length = $unpacked['length'];
+
+            if ($length < 4) {
+                throw new RecvFailedException(sprintf('Invalid package length: %d', $length));
+            }
+            $body = $this->readBytes($length);
+            if (in_array($body, [self::PING, self::PONG], true)) {
+                continue;
+            }
+
+            return $header . $body;
+        }
+    }
+
+    /**
+     * @throws Exception
+     */
+    private function readBytes(int $length): string
+    {
+        $buffer = '';
+
+        while (strlen($buffer) < $length) {
+            $read = [$this->client];
+            $write = null;
+            $except = null;
+
+            $selected = stream_select($read, $write, $except, $this->timeout);
+            if ($selected === false) {
+                throw new RuntimeException('Failed to select stream.');
+            }
+
+            if ($selected === 0) {
+                throw new RecvFailedException('Receive timeout.');
+            }
+
+            foreach ($read as $stream) {
+                $chunk = fread($stream, $length - strlen($buffer));
+
+                if ($chunk === false) {
+                    throw new RecvFailedException('Receive failed.');
+                }
+
+                if ($chunk === '' && feof($stream)) {
+                    throw new ConnectionException('Connection was closed.');
+                }
+
+                $buffer .= $chunk;
+            }
+        }
+
+        return $buffer;
+    }
+}

src/Transporter/StreamSocketTransporter.php

@@ -24,7 +24,7 @@
 class StreamSocketTransporter extends AbstractTransporter
 {
     /**
-     * @var resource|null
+     * @var null|resource
      */
     protected $client;