[rdkafka] Don't do unnecessary subscribe\unsubscribe on every receive call

Author: makasim

pkg/rdkafka/RdKafkaConsumer.php

@@ -82,7 +82,11 @@ public function getQueue()
      */
     public function receive($timeout = 0)
     {
-        $this->consumer->subscribe([$this->topic->getTopicName()]);
+        if (false == $this->subscribed) {
+            $this->consumer->subscribe([$this->topic->getTopicName()]);
+
+            $this->subscribed = true;
+        }
 
         $message = null;
         if ($timeout > 0) {
@@ -95,8 +99,6 @@ public function receive($timeout = 0)
             }
         }
 
-        $this->consumer->unsubscribe();
-
         return $message;
     }
 

pkg/rdkafka/RdKafkaContext.php

@@ -29,12 +29,18 @@ class RdKafkaContext implements PsrContext
      */
     private $producer;
 
+    /**
+     * @var KafkaConsumer[]
+     */
+    private $kafkaConsumers;
+
     /**
      * @param array $config
      */
     public function __construct(array $config)
     {
         $this->config = $config;
+        $this->kafkaConsumers = [];
 
         $this->setSerializer(new JsonSerializer());
     }
@@ -94,8 +100,10 @@ public function createConsumer(PsrDestination $destination)
     {
         InvalidDestinationException::assertDestinationInstanceOf($destination, RdKafkaTopic::class);
 
+        $this->kafkaConsumers[] = $kafkaConsumer = new KafkaConsumer($this->getConf());
+
         $consumer = new RdKafkaConsumer(
-            new KafkaConsumer($this->getConf()),
+            $kafkaConsumer,
             $this,
             $destination,
             $this->getSerializer()
@@ -113,6 +121,12 @@ public function createConsumer(PsrDestination $destination)
      */
     public function close()
     {
+        $kafkaConsumers = $this->kafkaConsumers;
+        $this->kafkaConsumers = [];
+
+        foreach ($kafkaConsumers as $kafkaConsumer) {
+            $kafkaConsumer->unsubscribe();
+        }
     }
 
     /**