added reconfigurable timeouts on a streamsocket

Author: transistive

src/connection/AConnection.php

@@ -71,4 +71,9 @@ public function getTimeout(): float
     {
         return $this->timeout;
     }
+
+    public function setTimeout(float $timeout): void
+    {
+        $this->timeout = $timeout;
+    }
 }

src/connection/IConnection.php

@@ -31,4 +31,6 @@ public function getIp(): string;
     public function getPort(): int;
 
     public function getTimeout(): float;
+
+    public function setTimeout(float $timeout): void;
 }

src/connection/StreamSocket.php

@@ -5,6 +5,8 @@
 
 use Bolt\Bolt;
 use Bolt\error\ConnectException;
+use function floor;
+use function stream_set_timeout;
 
 /**
  * Stream socket class
@@ -66,10 +68,7 @@ public function connect(): bool
             }
         }
 
-        $timeout = (int) floor($this->timeout);
-        if (!stream_set_timeout($this->stream, $timeout, (int) floor(($this->timeout - $timeout) * 1000000))) {
-            throw new ConnectException('Cannot set timeout on stream');
-        }
+        $this->setTimeoutOnStream();
 
         return true;
     }
@@ -120,4 +119,21 @@ public function disconnect()
             stream_socket_shutdown($this->stream, STREAM_SHUT_RDWR);
     }
 
+    public function setTimeout(float $timeout): void
+    {
+        parent::setTimeout($timeout);
+        $this->setTimeoutOnStream();
+    }
+
+    /**
+     * @return void
+     * @throws ConnectException
+     */
+    private function setTimeoutOnStream(): void
+    {
+        $timeout = (int)floor($this->timeout);
+        if (!stream_set_timeout($this->stream, $timeout, (int)floor(($this->timeout - $timeout) * 1000000))) {
+            throw new ConnectException('Cannot set timeout on stream');
+        }
+    }
 }

tests/connection/StreamSocketTest.php

@@ -5,8 +5,12 @@
 use Bolt\Bolt;
 use Bolt\connection\StreamSocket;
 use Bolt\error\ConnectException;
+use Bolt\error\MessageException;
+use Bolt\helpers\Auth;
+use Bolt\protocol\V4;
 use Exception;
 use PHPUnit\Framework\TestCase;
+use function microtime;
 
 final class StreamSocketTest extends TestCase
 {
@@ -18,7 +22,7 @@ public function testMillisecondTimeout(): void
         $socket = new StreamSocket($GLOBALS['NEO_HOST'] ?? '127.0.0.1', (int) ($GLOBALS['NEO_PORT'] ?? 7687), 1.5);
         $bolt = new Bolt($socket);
         $protocol = $bolt->build();
-        $protocol->init(\Bolt\helpers\Auth::basic($GLOBALS['NEO_USER'], $GLOBALS['NEO_PASS']));
+        $protocol->init(Auth::basic($GLOBALS['NEO_USER'], $GLOBALS['NEO_PASS']));
 
         $time = microtime(true);
         try {
@@ -42,7 +46,7 @@ public function testSecondsTimeout(): void
         $socket = new StreamSocket($GLOBALS['NEO_HOST'] ?? '127.0.0.1', (int) ($GLOBALS['NEO_PORT'] ?? 7687), 1);
         $bolt = new Bolt($socket);
         $protocol = $bolt->build();
-        $protocol->init(\Bolt\helpers\Auth::basic($GLOBALS['NEO_USER'], $GLOBALS['NEO_PASS']));
+        $protocol->init(Auth::basic($GLOBALS['NEO_USER'], $GLOBALS['NEO_PASS']));
 
         $time = microtime(true);
         try {
@@ -56,4 +60,48 @@ public function testSecondsTimeout(): void
             $this->assertEqualsWithDelta(1.0, $newTime - $time, 0.2);
         }
     }
+
+    public function testTimeoutRecoverAndReset(): void
+    {
+        $streamSocket = new StreamSocket($GLOBALS['NEO_HOST'] ?? '127.0.0.1', (int) ($GLOBALS['NEO_PORT'] ?? 7687), 1);
+        /** @var V4 $protocol */
+        $protocol = (new Bolt($streamSocket))->build();
+        $protocol->hello(Auth::basic($GLOBALS['NEO_USER'], $GLOBALS['NEO_PASS']));
+
+        $time = microtime(true);
+        try {
+            $protocol->run('FOREACH ( i IN range(1,10000) | 
+                MERGE (d:Day {day: i})
+            )');
+            $this->fail('No timeout error triggered');
+        } catch (ConnectException $e) {
+            $newTime = microtime(true);
+
+            $this->assertEqualsWithDelta(1.0, $newTime - $time, 0.2);
+        }
+
+        $streamSocket->setTimeout(100.0);
+        try {
+            $protocol->reset();
+        } catch (MessageException $e) {
+            echo $e->getMessage();
+            $protocol = (new Bolt($streamSocket))->build();
+            $protocol->hello(Auth::basic($GLOBALS['NEO_USER'], $GLOBALS['NEO_PASS']));
+        }
+        $this->assertTrue(true);
+
+        $streamSocket->setTimeout(1.0);
+
+        $time = microtime(true);
+        try {
+            $protocol->run('FOREACH ( i IN range(1,10000) | 
+                MERGE (d:Day {day: i})
+            )');
+            $this->fail('No timeout error triggered');
+        } catch (ConnectException $e) {
+            $newTime = microtime(true);
+
+            $this->assertEqualsWithDelta(1.0, $newTime - $time, 0.2);
+        }
+    }
 }