client reconnect on consumer timeout fix

basis-company · Feb 12, 2024 · b4ee0b6 · b4ee0b6
1 parent 8ea9329
commit b4ee0b6
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 4 deletions.
diff --git a/src/Client.php b/src/Client.php
@@ -229,7 +229,7 @@ public function setTimeout(float $value): self
     /**
      * @throws Throwable
      */
-    public function process(null|int|float $timeout = 0, bool $reply = true)
+    public function process(null|int|float $timeout = 0, bool $reply = true, bool $checkTimeout = true)
     {
         $this->lastDataReadFailureAt = null;
         $max = microtime(true) + $timeout;
@@ -238,7 +238,7 @@ public function process(null|int|float $timeout = 0, bool $reply = true)
         $iteration = 0;
         while (true) {
             try {
-                $line = $this->readLine(1024, "\r\n");
+                $line = $this->readLine(1024, "\r\n", $checkTimeout);
 
                 if ($line && ($this->ping || trim($line) != 'PONG')) {
                     break;

diff --git a/src/Consumer/Consumer.php b/src/Consumer/Consumer.php
@@ -143,7 +143,8 @@ public function handle(Closure $handler, Closure $emptyHandler = null, bool $ack
             foreach (range(1, $this->batch) as $_) {
                 $runtime->empty = true;
                 // expires request means that we should receive answer from stream
-                $this->client->process($this->expires ? PHP_INT_MAX : null, $ack);
+                // consumer timeout can be more that client connection timeout
+                $this->client->process($this->expires ? PHP_INT_MAX : null, $ack, false);
 
                 if ($runtime->empty) {
                     if ($emptyHandler) {

diff --git a/tests/Functional/StreamTest.php b/tests/Functional/StreamTest.php
@@ -116,7 +116,7 @@ public function testNoMessages()
         $this->called = false;
         $this->empty = false;
 
-        $stream = $this->getClient()->getApi()->getStream('no_messages');
+        $stream = $this->createClient(['reconnect' => false])->getApi()->getStream('no_messages');
         $stream->getConfiguration()->setSubjects(['cucumber']);
         $stream->create();
 
@@ -126,6 +126,7 @@ public function testNoMessages()
         $consumer->create()
             ->setDelay(0)
             ->setIterations(1)
+            ->setExpires(1)
             ->handle(function ($response) {
                 $this->called = $response;
             }, function () {