[kafka] fix tests.

makasim · makasim · commit 7e7396704b9a · 2018-01-04T23:56:35.000+02:00
diff --git a/pkg/rdkafka/Tests/RdKafkaConsumerTest.php b/pkg/rdkafka/Tests/RdKafkaConsumerTest.php
@@ -59,9 +59,34 @@ public function testShouldReceiveFromQueueAndReturnNullIfNoMessageInQueue()
             ->with(1000)
             ->willReturn($kafkaMessage)
         ;
+
+        $consumer = new RdKafkaConsumer(
+            $kafkaConsumer,
+            $this->createContextMock(),
+            $destination,
+            $this->createSerializerMock()
+        );
+
+        $this->assertNull($consumer->receive(1000));
+    }
+
+    public function testShouldSubscribeOnFirstReceiveOnly()
+    {
+        $destination = new RdKafkaTopic('dest');
+
+        $kafkaMessage = new Message();
+        $kafkaMessage->err = RD_KAFKA_RESP_ERR__TIMED_OUT;
+
+        $kafkaConsumer = $this->createKafkaConsumerMock();
         $kafkaConsumer
             ->expects($this->once())
-            ->method('unsubscribe')
+            ->method('subscribe')
+            ->with(['dest'])
+        ;
+        $kafkaConsumer
+            ->expects($this->any())
+            ->method('consume')
+            ->willReturn($kafkaMessage)
         ;
 
         $consumer = new RdKafkaConsumer(
@@ -71,7 +96,9 @@ public function testShouldReceiveFromQueueAndReturnNullIfNoMessageInQueue()
             $this->createSerializerMock()
         );
 
-        $this->assertNull($consumer->receive(1000));
+        $consumer->receive(1000);
+        $consumer->receive(1000);
+        $consumer->receive(1000);
     }
 
     public function testShouldReceiveFromQueueAndReturnMessageIfMessageInQueue()
@@ -96,10 +123,6 @@ public function testShouldReceiveFromQueueAndReturnMessageIfMessageInQueue()
             ->with(1000)
             ->willReturn($kafkaMessage)
         ;
-        $kafkaConsumer
-            ->expects($this->once())
-            ->method('unsubscribe')
-        ;
 
         $serializer = $this->createSerializerMock();
         $serializer
