Merge pull request #263 from php-enqueue/rdkafka-topic-default-config

[rdkafka] do not pass config if it was not set explisitly.

Author: makasim

Diff for: pkg/rdkafka/RdKafkaTopic.php

@@ -34,7 +34,6 @@ class RdKafkaTopic implements PsrTopic, PsrQueue
     public function __construct($name)
     {
         $this->name = $name;
-        $this->conf = new TopicConf();
     }
 
     /**
@@ -54,13 +53,21 @@ public function getQueueName()
     }
 
     /**
-     * @return TopicConf
+     * @return TopicConf|null
      */
     public function getConf()
     {
         return $this->conf;
     }
 
+    /**
+     * @param TopicConf|null $conf
+     */
+    public function setConf(TopicConf $conf = null)
+    {
+        $this->conf = $conf;
+    }
+
     /**
      * @return int
      */

Diff for: pkg/rdkafka/Tests/RdKafkaProducerTest.php

@@ -64,7 +64,7 @@ public function testShouldUseSerializerToEncodeMessageAndPutToExpectedTube()
         $kafkaProducer
             ->expects($this->once())
             ->method('newTopic')
-            ->with('theQueueName', $this->isInstanceOf(TopicConf::class))
+            ->with('theQueueName')
             ->willReturn($kafkaTopic)
         ;
 
@@ -81,6 +81,74 @@ public function testShouldUseSerializerToEncodeMessageAndPutToExpectedTube()
         $producer->send(new RdKafkaTopic('theQueueName'), $message);
     }
 
+    public function testShouldPassNullAsTopicConfigIfNotSetOnTopic()
+    {
+        // guard
+        $kafkaTopic = $this->createKafkaTopicMock();
+        $kafkaTopic
+            ->expects($this->once())
+            ->method('produce')
+        ;
+
+        $kafkaProducer = $this->createKafkaProducerMock();
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('newTopic')
+            ->with('theQueueName', null)
+            ->willReturn($kafkaTopic)
+        ;
+
+        $serializer = $this->createSerializerMock();
+        $serializer
+            ->expects($this->once())
+            ->method('toString')
+            ->willReturn('aSerializedMessage')
+        ;
+
+        $producer = new RdKafkaProducer($kafkaProducer, $serializer);
+
+        $topic = new RdKafkaTopic('theQueueName');
+
+        // guard
+        $this->assertNull($topic->getConf());
+
+        $producer->send($topic, new RdKafkaMessage());
+    }
+
+    public function testShouldPassCustomConfAsTopicConfigIfSetOnTopic()
+    {
+        $conf = new TopicConf();
+
+        // guard
+        $kafkaTopic = $this->createKafkaTopicMock();
+        $kafkaTopic
+            ->expects($this->once())
+            ->method('produce')
+        ;
+
+        $kafkaProducer = $this->createKafkaProducerMock();
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('newTopic')
+            ->with('theQueueName', $this->identicalTo($conf))
+            ->willReturn($kafkaTopic)
+        ;
+
+        $serializer = $this->createSerializerMock();
+        $serializer
+            ->expects($this->once())
+            ->method('toString')
+            ->willReturn('aSerializedMessage')
+        ;
+
+        $producer = new RdKafkaProducer($kafkaProducer, $serializer);
+
+        $topic = new RdKafkaTopic('theQueueName');
+        $topic->setConf($conf);
+
+        $producer->send($topic, new RdKafkaMessage());
+    }
+
     public function testShouldAllowGetPreviouslySetSerializer()
     {
         $producer = new RdKafkaProducer($this->createKafkaProducerMock(), $this->createSerializerMock());

Diff for: pkg/rdkafka/Tests/RdKafkaTopicTest.php

@@ -27,10 +27,20 @@ public function testCouldSetGetKey()
         $this->assertSame('key', $topic->getKey());
     }
 
-    public function testShouldReturnConfInstance()
+    public function testShouldReturnNullAsConfIfNotSet()
     {
         $topic = new RdKafkaTopic('topic');
 
-        $this->assertInstanceOf(TopicConf::class, $topic->getConf());
+        $this->assertNull($topic->getConf());
+    }
+
+    public function testShouldAllowGetPreviouslySetConf()
+    {
+        $topic = new RdKafkaTopic('topic');
+
+        $conf = new TopicConf();
+        $topic->setConf($conf);
+
+        $this->assertSame($conf, $topic->getConf());
     }
 }