Create singleton to avoid consumer duplication

Author: dirk39

pkg/rdkafka/RdKafkaContext.php

@@ -45,6 +45,11 @@ class RdKafkaContext implements Context
      */
     private $kafkaConsumers;
 
+    /**
+     * @var RdKafkaConsumer[]
+     */
+    private $rdKafkaConsumers;
+
     /**
      * @param array $config
      */
@@ -102,20 +107,26 @@ public function createConsumer(Destination $destination): Consumer
     {
         InvalidDestinationException::assertDestinationInstanceOf($destination, RdKafkaTopic::class);
 
-        $this->kafkaConsumers[] = $kafkaConsumer = new KafkaConsumer($this->getConf());
+        $queueName = $destination->getQueueName();
 
-        $consumer = new RdKafkaConsumer(
-            $kafkaConsumer,
-            $this,
-            $destination,
-            $this->getSerializer()
-        );
+        if (!isset($this->rdKafkaConsumers[$queueName])) {
+            $this->kafkaConsumers[] = $kafkaConsumer = new KafkaConsumer($this->getConf());
+
+            $consumer = new RdKafkaConsumer(
+                $kafkaConsumer,
+                $this,
+                $destination,
+                $this->getSerializer()
+            );
+
+            if (isset($this->config['commit_async'])) {
+                $consumer->setCommitAsync($this->config['commit_async']);
+            }
 
-        if (isset($this->config['commit_async'])) {
-            $consumer->setCommitAsync($this->config['commit_async']);
+            $this->rdKafkaConsumers[$queueName] = $consumer;
         }
 
-        return $consumer;
+        return $this->rdKafkaConsumers[$queueName];
     }
 
     public function close(): void

pkg/rdkafka/Tests/RdKafkaContextTest.php

@@ -69,4 +69,27 @@ public function testShouldInjectItsSerializerToConsumer()
 
         $this->assertSame($context->getSerializer(), $producer->getSerializer());
     }
+
+    public function testShouldNotCreateConsumerTwice()
+    {
+        $context = new RdKafkaContext([]);
+        $queue = $context->createQueue('aQueue');
+
+        $consumer = $context->createConsumer($queue);
+        $consumer2 = $context->createConsumer($queue);
+
+        $this->assertSame($consumer, $consumer2);
+    }
+
+    public function testShouldCreateTwoConsumers()
+    {
+        $context = new RdKafkaContext([]);
+        $queueA = $context->createQueue('aQueue');
+        $queueB = $context->createQueue('aQueue');
+
+        $consumer = $context->createConsumer($queueA);
+        $consumer2 = $context->createConsumer($queueB);
+
+        $this->assertNotSame($consumer, $consumer2);
+    }
 }