feat: Refactor code around specialty handling of SocketRelay and add …

…more test coverage
roadrunner-php · L3tum · Feb 5, 2024 · Feb 5, 2024 · Feb 5, 2024 · Feb 5, 2024
commit bb190ce8f26e3568703aee79e5fd9133ecde5454
diff --git a/src/ConnectedRelayInterface.php b/src/ConnectedRelayInterface.php
@@ -0,0 +1,30 @@
+<?php
+
+namespace Spiral\Goridge;
+
+use Spiral\Goridge\Exception\RelayException;
+
+/**
+ * This interface describes a Relay that explictily establishes a connection.
+ * That connection can also be re-established on the fly (in comparison to StreamRelay, which relies on the existence of the streams).
+ * The object is also clonable, i.e. supports cloning without data errors due to shared state.
+ */
+interface ConnectedRelayInterface
+{
+    /**
+     * Returns true if the underlying connection is already established
+     */
+    public function isConnected(): bool;
+
+    /**
+     * Establishes the underlying connection and returns true on success, false on failure, or throws an exception in case of an error.
+     *
+     * @throws RelayException
+     */
+    public function connect(): bool;
+
+    /**
+     * Closes the underlying connection.
+     */
+    public function close(): void;
+}
diff --git a/src/MultiRelayHelper.php b/src/MultiRelayHelper.php
@@ -89,7 +89,7 @@ public static function findRelayWithMessage(array $relays, int $timeoutInMicrose
     }
 
     /**
-     * @param array<array-key, RelayInterface> $relays
+     * @param array<array-key, ConnectedRelayInterface> $relays
      * @return array-key[]|false
      * @internal
      * Returns either
@@ -98,14 +98,14 @@ public static function findRelayWithMessage(array $relays, int $timeoutInMicrose
      */
     public static function checkConnected(array $relays): array|false
     {
-        if (count($relays) === 0 || !$relays[array_key_first($relays)] instanceof SocketRelay) {
+        if (count($relays) === 0) {
             return false;
         }
 
         $keysNotConnected = [];
         foreach ($relays as $key => $relay) {
-            assert($relay instanceof SocketRelay);
-            if ($relay->socket === null) {
+            assert($relay instanceof ConnectedRelayInterface);
+            if (!$relay->isConnected()) {
                 $relay->connect();
                 $keysNotConnected[] = $key;
             }

diff --git a/src/RPC/MultiRPC.php b/src/RPC/MultiRPC.php
@@ -4,6 +4,7 @@
 
 namespace Spiral\Goridge\RPC;
 
+use Spiral\Goridge\ConnectedRelayInterface;
 use Spiral\Goridge\Exception\RelayException;
 use Spiral\Goridge\Exception\TransportException;
 use Spiral\Goridge\Frame;
@@ -23,18 +24,21 @@ class MultiRPC extends AbstractRPC implements AsyncRPCInterface
     private const DEFAULT_BUFFER_THRESHOLD = 10_000;
 
     /**
-     * @var array<int, RelayInterface>
+     * @var array<int, ConnectedRelayInterface>
      */
     private array $freeRelays = [];
 
     /**
-     * Occupied Relays alone is a map of seq to relay to make removal easier once a response is received.
-     * @var array<positive-int, RelayInterface>
+     * Occupied Relays is a map of seq to relay to make removal easier once a response is received.
+     * @var array<positive-int, ConnectedRelayInterface>
      */
     private array $occupiedRelays = [];
 
     /**
-     * @var array<positive-int, RelayInterface>
+     * A map of seq to relay to use for decodeResponse().
+     * Technically the relay there is only needed in case of an error.
+     *
+     * @var array<positive-int, ConnectedRelayInterface>
      */
     private array $seqToRelayMap = [];
 
@@ -46,6 +50,9 @@ class MultiRPC extends AbstractRPC implements AsyncRPCInterface
      */
     private array $asyncResponseBuffer = [];
 
+    /**
+     * The threshold after which the asyncResponseBuffer is flushed of all entries.
+     */
     private int $asyncBufferThreshold = self::DEFAULT_BUFFER_THRESHOLD;
 
     /**
@@ -56,9 +63,19 @@ public function __construct(
         int $asyncBufferThreshold = self::DEFAULT_BUFFER_THRESHOLD,
         CodecInterface $codec = new JsonCodec()
     ) {
+        if (count($relays) === 0) {
+            throw new RPCException("MultiRPC needs at least one relay. Zero provided.");
+        }
+
         foreach ($relays as $relay) {
-            if (!($relay instanceof SocketRelay)) {
-                throw new RPCException("MultiRPC can only be used with SocketRelay");
+            if (!($relay instanceof ConnectedRelayInterface)) {
+                throw new RPCException(
+                    sprintf(
+                        "MultiRPC can only be used with relays implementing the %s, such as %s",
+                        ConnectedRelayInterface::class,
+                        SocketRelay::class
+                    )
+                );
             }
         }
 
@@ -106,22 +123,12 @@ public static function create(
     }
 
     /**
-     * Force-connects all SocketRelays.
-     * Does nothing if no SocketRelay.
+     * Force-connects all relays.
      * @throws RelayException
      */
     public function preConnectRelays(): void
     {
-        if (count($this->freeRelays) === 0) {
-            return;
-        }
-
-        if (!$this->freeRelays[0] instanceof SocketRelay) {
-            return;
-        }
-
         foreach ($this->freeRelays as $relay) {
-            assert($relay instanceof SocketRelay);
             // Force connect
             $relay->connect();
         }
@@ -370,14 +377,11 @@ private function ensureFreeRelayAvailable(): int
     /**
      * Gets a response from the relay, blocking for it if necessary, with some error handling in regards to mismatched seq
      *
-     * @param RelayInterface $relay
      * @param positive-int $expectedSeq
-     * @param bool $onlySaveResponseInCaseOfMismatchedSeq
-     * @return Frame
      */
-    private function getResponseFromRelay(RelayInterface $relay, int $expectedSeq, bool $onlySaveResponseInCaseOfMismatchedSeq = false): Frame
+    private function getResponseFromRelay(ConnectedRelayInterface $relay, int $expectedSeq, bool $onlySaveResponseInCaseOfMismatchedSeq = false): Frame
     {
-        if ($relay instanceof SocketRelay && !$relay->isConnected()) {
+        if (!$relay->isConnected()) {
             throw new TransportException("Unable to read payload from the stream");
         }
 

diff --git a/src/SocketRelay.php b/src/SocketRelay.php
@@ -24,7 +24,7 @@
  *
  * @psalm-suppress DeprecatedInterface
  */
-class SocketRelay extends Relay implements Stringable
+class SocketRelay extends Relay implements Stringable, ConnectedRelayInterface
 {
     final public const RECONNECT_RETRIES = 10;
     final public const RECONNECT_TIMEOUT = 100;

diff --git a/tests/Goridge/MsgPackMultiRPCTest.php b/tests/Goridge/MsgPackMultiRPCTest.php
@@ -43,8 +43,8 @@ public function testJsonExceptionNotThrownWithIgnoreResponse(): void
     /**
      * @return MultiRPC
      */
-    protected function makeRPC(): MultiRPC
+    protected function makeRPC(int $count = 10): MultiRPC
     {
-        return parent::makeRPC()->withCodec(new MsgpackCodec());
+        return parent::makeRPC($count)->withCodec(new MsgpackCodec());
     }
 }
diff --git a/tests/Goridge/MultiRPC.php b/tests/Goridge/MultiRPC.php
@@ -7,6 +7,7 @@
 use PHPUnit\Framework\TestCase;
 use ReflectionMethod;
 use ReflectionProperty;
+use Spiral\Goridge\ConnectedRelayInterface;
 use Spiral\Goridge\Exception\TransportException;
 use Spiral\Goridge\RelayInterface;
 use Spiral\Goridge\RPC\Codec\MsgpackCodec;
@@ -659,6 +660,7 @@ public function testHandleRelayDisconnectWithPressureGetResponses(): void
     public function testHandleCloneCorrectly(): void
     {
         $this->rpc->preConnectRelays();
+        $this->rpc->callIgnoreResponse('Service.Ping', 'ping');
         $clonedRpc = $this->rpc->withCodec(new MsgpackCodec());
         for ($i = 0; $i < 50; $i++) {
             $clonedRpc->callIgnoreResponse('Service.Ping', 'ping');
@@ -684,12 +686,92 @@ public function testHandleCloneCorrectly(): void
         $this->assertSame('pong', $this->rpc->call('Service.Ping', 'ping'));
     }
 
-    public function testAllowsOnlySockets(): void{
+    public function testAllowsOnlySockets(): void
+    {
         $this->expectException(RPCException::class);
-        $this->expectExceptionMessage("MultiRPC can only be used with SocketRelay");
+        $this->expectExceptionMessage(
+            sprintf(
+                "MultiRPC can only be used with relays implementing the %s, such as %s",
+                ConnectedRelayInterface::class,
+                SocketRelay::class
+            )
+        );
         $this->rpc = new GoridgeMultiRPC([new StreamRelay(STDIN, STDOUT)]);
     }
 
+    public function testNeedsAtLeastOne(): void
+    {
+        $this->expectException(RPCException::class);
+        $this->expectExceptionMessage("MultiRPC needs at least one relay. Zero provided.");
+        $this->rpc = new GoridgeMultiRPC([]);
+    }
+
+    public function testChecksIfResponseIsInRelay(): void
+    {
+        $id = $this->rpc->callAsync('Service.Ping', 'ping');
+        // Wait a bit
+        usleep(100 * 1000);
+
+        $this->assertTrue($this->rpc->hasResponse($id));
+    }
+
+    public function testChecksIfResponseIsInBuffer(): void
+    {
+        $id = $this->rpc->callAsync('Service.Ping', 'ping');
+        // Wait a bit
+        usleep(100 * 1000);
+        $this->forceFlushRpc($this->rpc);
+
+        $this->assertTrue($this->rpc->hasResponse($id));
+    }
+
+    public function testChecksIfResponseIsNotReceivedYet(): void
+    {
+        $id = $this->rpc->callAsync('Service.Ping', 'ping');
+        $this->assertFalse($this->rpc->hasResponse($id));
+    }
+
+    public function testChecksMultipleResponses(): void
+    {
+        $ids = [];
+        $ids[] = $this->rpc->callAsync('Service.Ping', 'ping');
+        $this->forceFlushRpc($this->rpc);
+        $ids[] = $this->rpc->callAsync('Service.Ping', 'ping');
+        usleep(100 * 1000);
+        $ids[] = $this->rpc->callAsync('Service.Ping', 'ping');
+        $responses = $this->rpc->hasResponses($ids);
+        $this->assertContains($ids[0], $responses);
+        $this->assertContains($ids[1], $responses);
+        $this->assertNotContains($ids[2], $responses);
+    }
+
+    public function testHasResponsesReturnsEmptyArrayWhenNoResponses(): void
+    {
+        $id = $this->rpc->callAsync('Service.Ping', 'ping');
+        $this->assertEmpty($this->rpc->hasResponses([$id]));
+    }
+
+    public function testGetResponsesReturnsWhenNoRelaysAvailableToAvoidInfiniteLoop(): void
+    {
+        // occupiedRelays is already empty
+        $rpc = $this->makeRPC();
+        $property = new ReflectionProperty(GoridgeMultiRPC::class, 'freeRelays');
+        $property->setValue($rpc, []);
+        $this->expectException(RPCException::class);
+        $this->expectExceptionMessage("No relays available at all");
+        $rpc->call('Service.Ping', 'ping');
+    }
+
+    public function testMultiRPCIsUsableWithOneRelay(): void
+    {
+        $rpc = $this->makeRPC(1);
+        $rpc->callIgnoreResponse('Service.Ping', 'ping');
+        $rpc->callIgnoreResponse('Service.SleepEcho', 'Hello');
+        $id = $rpc->callAsync('Service.Ping', 'ping');
+        $rpc->callIgnoreResponse('Service.Echo', 'Hello');
+        $this->assertSame('pong', $rpc->getResponse($id));
+    }
+
     protected function setUp(): void
     {
         $this->rpc = $this->makeRPC();
@@ -698,13 +780,12 @@ protected function setUp(): void
     /**
      * @return GoridgeMultiRPC
      */
-    protected function makeRPC(): GoridgeMultiRPC
+    protected function makeRPC(int $count = 10): GoridgeMultiRPC
     {
-        $relays = [];
-        for ($i = 0; $i < 10; $i++) {
-            $relays[] = $this->makeRelay();
-        }
-        return new GoridgeMultiRPC($relays);
+        $type = self::SOCK_TYPE->value;
+        $address = self::SOCK_ADDR;
+        $port = self::SOCK_PORT;
+        return GoridgeMultiRPC::create("$type://$address:$port", $count);
     }
 
     /**

diff --git a/tests/Goridge/MultiRelayHelperTest.php b/tests/Goridge/MultiRelayHelperTest.php
@@ -0,0 +1,17 @@
+<?php
+
+namespace Goridge;
+
+use PHPUnit\Framework\TestCase;
+use Spiral\Goridge\MultiRelayHelper;
+use Spiral\Goridge\StreamRelay;
+
+class MultiRelayHelperTest extends TestCase
+{
+    public function testSupportsStreamRelay(): void
+    {
+        $relays = [new StreamRelay(STDIN, STDOUT), new StreamRelay(STDIN, STDERR)];
+        // No message available on STDIN, aka a read would block, so this returns false
+        $this->assertFalse(MultiRelayHelper::findRelayWithMessage($relays));
+    }
+}