Merge pull request #575 from Steveb-p/Steveb-p-patch-1

[rdkafka] Backport changes to topic subscription
php-enqueue · Oct 22, 2018 · 0b091aa · 0b091aa
2 parents 79f1821 + 345b6ac
commit 0b091aa
Showing 1 changed file with 9 additions and 5 deletions.
diff --git a/pkg/rdkafka/RdKafkaConsumer.php b/pkg/rdkafka/RdKafkaConsumer.php
@@ -99,11 +99,15 @@ public function getQueue()
     public function receive($timeout = 0)
     {
         if (false == $this->subscribed) {
-            $this->consumer->assign([new TopicPartition(
-                $this->getQueue()->getQueueName(),
-                $this->getQueue()->getPartition(),
-                $this->offset
-            )]);
+            if (null === $this->offset) {
+                $this->consumer->subscribe([$this->getQueue()->getQueueName()]);
+            } else {
+                $this->consumer->assign([new TopicPartition(
+                    $this->getQueue()->getQueueName(),
+                    $this->getQueue()->getPartition(),
+                    $this->offset
+                )]);
+            }
 
             $this->subscribed = true;
         }