[1.7.x] Fix serialize

src/Producers/Producer.php

@@ -53,6 +53,8 @@ public function produce(KafkaProducerMessage $message): bool
     {
         $topic = $this->producer->newTopic($this->topic);
 
+        $message = clone $message;
+
         $message = $this->serializer->serialize($message);
 
         $this->produceMessage($topic, $message);
@@ -75,6 +77,8 @@ public function produceBatch(MessageBatch $messageBatch): int
 
         $produced = 0;
         foreach ($messagesIterator as $message) {
+            $message = clone $message;
+
             $message = $this->serializer->serialize($message);
 
             $this->produceMessage($topic, $message);

tests/LaravelKafkaTestCase.php

@@ -11,6 +11,7 @@
 use RdKafka\KafkaConsumer;
 use RdKafka\Message;
 use RdKafka\Producer as KafkaProducer;
+use RdKafka\Conf;
 
 class LaravelKafkaTestCase extends Orchestra
 {
@@ -57,10 +58,25 @@ protected function mockProducer()
         $this->app->bind(Producer::class, function () use ($mockedProducer) {
             return $mockedProducer->getMock();
         });
+        $this->mockKafkaProducer();
+    }
+
+    protected function mockKafkaProducer()
+    {
+        // We have to get a topic object as a valid response for the mock
+        // We stub out this code here to achieve that
+        $conf = new Conf();
+        $conf->set('log_level', 0);
+        $kafka = new KafkaProducer($conf);
+        $topic = $kafka->newTopic('test-topic');
 
         $mockedKafkaProducer = m::mock(KafkaProducer::class)
             ->shouldReceive('flush')
             ->andReturn(RD_KAFKA_RESP_ERR_NO_ERROR)
+            ->shouldReceive('newTopic')
+            ->andReturn($topic)
+            ->shouldReceive('poll')
+            ->andReturn(RD_KAFKA_RESP_ERR_NO_ERROR)
             ->getMock();
 
         $this->app->bind(KafkaProducer::class, function () use ($mockedKafkaProducer) {

tests/Producers/ProducerTest.php

@@ -0,0 +1,47 @@
+<?php
+
+namespace Junges\Kafka\Tests\Producers;
+
+use Junges\Kafka\Config\Config;
+use Junges\Kafka\Message\Serializers\JsonSerializer;
+use Junges\Kafka\Producers\MessageBatch;
+use Junges\Kafka\Producers\Producer;
+use Junges\Kafka\Tests\LaravelKafkaTestCase;
+use Junges\Kafka\Message\Message;
+
+class ProducerTest extends LaravelKafkaTestCase
+{
+    public function test_it_does_not_double_serialize_a_message_when_using_json_serializer()
+    {
+        $this->mockKafkaProducer();
+        $producer = new Producer(new Config('broker', ['test-topic']), 'test-topic', new JsonSerializer());
+        $payload = ['key' => 'value'];
+
+        $message = new Message(
+            body: $payload,
+        );
+        $producer->produce($message);
+        $producer->produce($message);
+
+        $this->assertSame($payload, $message->getBody());
+    }
+
+    public function test_it_does_not_double_serialize_a_batch_message_when_using_json_serializer()
+    {
+        $this->mockKafkaProducer();
+        $producer = new Producer(new Config('broker', ['test-topic']), 'test-topic', new JsonSerializer());
+        $payload = ['key' => 'value'];
+
+        $message = new Message(
+            body: $payload,
+        );
+
+        $messageBatch = new MessageBatch();
+        $messageBatch->push($message);
+
+        $producer->produceBatch($messageBatch);
+        $producer->produceBatch($messageBatch);
+
+        $this->assertSame($payload, $message->getBody());
+    }
+}