Merge pull request #254 from flix-tech/feature-enable-key-serialization

[RdKafka] Enable serializers to serialize message keys
php-enqueue · Nov 5, 2017 · 781eb0d · 781eb0d
2 parents 53040ff + 58c7d4f
commit 781eb0d
Show file tree

Hide file tree

Showing 2 changed files with 40 additions and 1 deletion.
diff --git a/pkg/rdkafka/RdKafkaProducer.php b/pkg/rdkafka/RdKafkaProducer.php
@@ -41,8 +41,8 @@ public function send(PsrDestination $destination, PsrMessage $message)
         InvalidMessageException::assertMessageInstanceOf($message, RdKafkaMessage::class);
 
         $partition = $message->getPartition() ?: $destination->getPartition() ?: RD_KAFKA_PARTITION_UA;
-        $key = $message->getKey() ?: $destination->getKey() ?: null;
         $payload = $this->serializer->toString($message);
+        $key = $message->getKey() ?: $destination->getKey() ?: null;
 
         $topic = $this->producer->newTopic($destination->getTopicName(), $destination->getConf());
         $topic->produce($partition, 0 /* must be 0 */, $payload, $key);

diff --git a/pkg/rdkafka/Tests/RdKafkaProducerTest.php b/pkg/rdkafka/Tests/RdKafkaProducerTest.php
@@ -95,6 +95,45 @@ public function testShouldAllowGetPreviouslySetSerializer()
         $this->assertSame($expectedSerializer, $producer->getSerializer());
     }
 
+    public function testShouldAllowSerializersToSerializeKeys()
+    {
+        $message = new RdKafkaMessage('theBody', ['foo' => 'fooVal'], ['bar' => 'barVal']);
+        $message->setKey('key');
+
+        $kafkaTopic = $this->createKafkaTopicMock();
+        $kafkaTopic
+            ->expects($this->once())
+            ->method('produce')
+            ->with(
+                RD_KAFKA_PARTITION_UA,
+                0,
+                'theSerializedMessage',
+                'theSerializedKey'
+            )
+        ;
+
+        $kafkaProducer = $this->createKafkaProducerMock();
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('newTopic')
+            ->willReturn($kafkaTopic)
+        ;
+
+        $serializer = $this->createSerializerMock();
+        $serializer
+            ->expects($this->once())
+            ->method('toString')
+            ->willReturnCallback(function () use ($message) {
+                $message->setKey('theSerializedKey');
+
+                return 'theSerializedMessage';
+            })
+        ;
+
+        $producer = new RdKafkaProducer($kafkaProducer, $serializer);
+        $producer->send(new RdKafkaTopic('theQueueName'), $message);
+    }
+
     /**
      * @return \PHPUnit_Framework_MockObject_MockObject|ProducerTopic
      */