From e31d04d88e8df7fd423bfabd8c7638e37efeb3c3 Mon Sep 17 00:00:00 2001
From: Maksim Kotlyar <kotlyar.maksim@gmail.com>
Date: Fri, 10 Nov 2017 16:55:46 +0200
Subject: [PATCH] [rdkafka] do not pass config if it was not set explisitly.

---
 pkg/rdkafka/RdKafkaTopic.php              | 11 +++-
 pkg/rdkafka/Tests/RdKafkaProducerTest.php | 70 ++++++++++++++++++++++-
 pkg/rdkafka/Tests/RdKafkaTopicTest.php    | 14 ++++-
 3 files changed, 90 insertions(+), 5 deletions(-)

diff --git a/pkg/rdkafka/RdKafkaTopic.php b/pkg/rdkafka/RdKafkaTopic.php
index 9aea638ee..6faf12f8c 100644
--- a/pkg/rdkafka/RdKafkaTopic.php
+++ b/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 --git a/pkg/rdkafka/Tests/RdKafkaProducerTest.php b/pkg/rdkafka/Tests/RdKafkaProducerTest.php
index 6ac08bbab..1c076e498 100644
--- a/pkg/rdkafka/Tests/RdKafkaProducerTest.php
+++ b/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 --git a/pkg/rdkafka/Tests/RdKafkaTopicTest.php b/pkg/rdkafka/Tests/RdKafkaTopicTest.php
index f81e4cd38..5ed22885a 100644
--- a/pkg/rdkafka/Tests/RdKafkaTopicTest.php
+++ b/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());
     }
 }