[1.6.x] Fix serialize (#92)

* 1.6 patch - Ensure json serialize does not double encode messages
* code clean

src/Producers/Producer.php

@@ -51,6 +51,8 @@ public function produce(KafkaProducerMessage $message): bool
     {
         $topic = $this->producer->newTopic($this->topic);
 
+        $message = clone $message;
+
         $message = $this->serializer->serialize($message);
 
         if (method_exists($topic, 'producev')) {

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
 {
@@ -58,9 +59,24 @@ protected function mockProducer()
             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,27 @@
+<?php
+
+namespace Junges\Kafka\Tests\Producers;
+
+use Junges\Kafka\Config\Config;
+use Junges\Kafka\Message\Serializers\JsonSerializer;
+use Junges\Kafka\Producers\Producer;
+use Junges\Kafka\Tests\LaravelKafkaTestCase;
+use Junges\Kafka\Message\Message;
+
+class ProducerTest extends LaravelKafkaTestCase
+{
+    public function testItDoesNotDoubleSerializeMessageWhenUsingJsonSerializer()
+    {
+        $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());
+    }
+}