New message transformation concept

pkg/rdkafka/ConsumeMessageTransformer.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\RdKafka;
+
+interface ConsumeMessageTransformer
+{
+    public function transformConsumeMessage(RdKafkaMessage $message);
+}

pkg/rdkafka/JsonSerializer.php

@@ -4,6 +4,9 @@
 
 namespace Enqueue\RdKafka;
 
+/**
+ * @deprecated Use ProduceMessageTransformer and/or ConsumeMessageTransformer instead
+ */
 class JsonSerializer implements Serializer
 {
     public function toString(RdKafkaMessage $message): string

pkg/rdkafka/NullMessageTransformer.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\RdKafka;
+
+final class NullMessageTransformer implements ConsumeMessageTransformer, ProduceMessageTransformer
+{
+
+    public function transformConsumeMessage(RdKafkaMessage $message)
+    {
+        return;
+    }
+
+    public function transformProduceMessage(RdKafkaMessage $message)
+    {
+        return;
+    }
+}

pkg/rdkafka/ProduceMessageTransformer.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\RdKafka;
+
+interface ProduceMessageTransformer
+{
+    public function transformProduceMessage(RdKafkaMessage $message);
+}

pkg/rdkafka/RdKafkaConsumer.php

@@ -45,6 +45,11 @@ class RdKafkaConsumer implements Consumer
      */
     private $offset;
 
+    /**
+     * @var ConsumeMessageTransformer
+     */
+    private $consumeMessageTransformer;
+
     public function __construct(KafkaConsumer $consumer, RdKafkaContext $context, RdKafkaTopic $topic, Serializer $serializer)
     {
         $this->consumer = $consumer;
@@ -164,7 +169,12 @@ private function doReceive(int $timeout): ?RdKafkaMessage
             case RD_KAFKA_RESP_ERR__TIMED_OUT:
                 break;
             case RD_KAFKA_RESP_ERR_NO_ERROR:
-                $message = $this->serializer->toMessage($kafkaMessage->payload);
+                if (null !== $this->serializer) {
+                    $message = $this->serializer->toMessage($kafkaMessage->payload);
+                } else {
+                    $message = new RdKafkaMessage($kafkaMessage->payload);
+                }
+
                 $message->setKey($kafkaMessage->key);
                 $message->setPartition($kafkaMessage->partition);
                 $message->setKafkaMessage($kafkaMessage);
@@ -175,6 +185,8 @@ private function doReceive(int $timeout): ?RdKafkaMessage
                     $message->setHeaders(array_merge($message->getHeaders(), $kafkaMessage->headers));
                 }
 
+                $this->consumeMessageTransformer->transformConsumeMessage($message);
+
                 return $message;
             default:
                 throw new \LogicException($kafkaMessage->errstr(), $kafkaMessage->err);
@@ -183,4 +195,9 @@ private function doReceive(int $timeout): ?RdKafkaMessage
 
         return null;
     }
+
+    public function setMessageTransformer(ConsumeMessageTransformer $messageTransformer): void
+    {
+        $this->consumeMessageTransformer = $messageTransformer;
+    }
 }

pkg/rdkafka/RdKafkaContext.php

@@ -50,6 +50,16 @@ class RdKafkaContext implements Context
      */
     private $rdKafkaConsumers;
 
+    /**
+     * @var ConsumeMessageTransformer
+     */
+    private $consumeMessageTransformer;
+
+    /**
+     * @var ProduceMessageTransformer
+     */
+    private $produceMessageTransformer;
+
     /**
      * @param array $config
      */
@@ -60,6 +70,11 @@ public function __construct(array $config)
         $this->rdKafkaConsumers = [];
 
         $this->setSerializer(new JsonSerializer());
+
+        $messageTransformer = new NullMessageTransformer();
+
+        $this->consumeMessageTransformer = $messageTransformer;
+        $this->produceMessageTransformer = $messageTransformer;
     }
 
     /**
@@ -96,7 +111,11 @@ public function createTemporaryQueue(): Queue
      */
     public function createProducer(): Producer
     {
-        return new RdKafkaProducer($this->getProducer(), $this->getSerializer());
+        $producer = new RdKafkaProducer($this->getProducer(), $this->getSerializer());
+
+        $producer->setMessageTransformer($this->produceMessageTransformer);
+
+        return $producer;
     }
 
     /**
@@ -120,6 +139,8 @@ public function createConsumer(Destination $destination): Consumer
                 $this->getSerializer()
             );
 
+            $consumer->setMessageTransformer($this->consumeMessageTransformer);
+
             if (isset($this->config['commit_async'])) {
                 $consumer->setCommitAsync($this->config['commit_async']);
             }

pkg/rdkafka/RdKafkaProducer.php

@@ -21,6 +21,11 @@ class RdKafkaProducer implements Producer
      */
     private $producer;
 
+    /**
+     * @var ProduceMessageTransformer
+     */
+    private $produceMessageTransformer;
+
     public function __construct(VendorProducer $producer, Serializer $serializer)
     {
         $this->producer = $producer;
@@ -38,7 +43,15 @@ public function send(Destination $destination, Message $message): void
         InvalidMessageException::assertMessageInstanceOf($message, RdKafkaMessage::class);
 
         $partition = $message->getPartition() ?: $destination->getPartition() ?: RD_KAFKA_PARTITION_UA;
-        $payload = $this->serializer->toString($message);
+
+        if (null !== $this->serializer) {
+            $payload = $this->serializer->toString($message);
+        } else {
+            $payload = $message->getBody();
+        }
+
+        $this->produceMessageTransformer->transformProduceMessage($message);
+
         $key = $message->getKey() ?: $destination->getKey() ?: null;
 
         $topic = $this->producer->newTopic($destination->getTopicName(), $destination->getConf());
@@ -111,4 +124,9 @@ public function getTimeToLive(): ?int
     {
         return null;
     }
+
+    public function setMessageTransformer(ProduceMessageTransformer $messageTransformer): void
+    {
+        $this->produceMessageTransformer = $messageTransformer;
+    }
 }

pkg/rdkafka/Serializer.php

@@ -4,6 +4,9 @@
 
 namespace Enqueue\RdKafka;
 
+/**
+ * @deprecated Use ProduceMessageTransformer and/or ConsumeMessageTransformer instead
+ */
 interface Serializer
 {
     public function toString(RdKafkaMessage $message): string;

pkg/rdkafka/SerializerAwareTrait.php

@@ -7,20 +7,20 @@
 trait SerializerAwareTrait
 {
     /**
-     * @var Serializer
+     * @var Serializer|null
      */
     private $serializer;
 
     /**
-     * @param Serializer $serializer
+     * @param Serializer|null $serializer
      */
-    public function setSerializer(Serializer $serializer)
+    public function setSerializer(?Serializer $serializer)
     {
         $this->serializer = $serializer;
     }
 
     /**
-     * @return Serializer
+     * @return Serializer|null
      */
     public function getSerializer()
     {