feat: Make serializer configurable via YAML configuration

- Add support for configuring serializers through YAML in RdKafkaContext
- Allow serializer specification as a class name, array with options, or instance

Author: lenchester

pkg/rdkafka/RdKafkaContext.php

@@ -16,6 +16,7 @@
 use Interop\Queue\Queue;
 use Interop\Queue\SubscriptionConsumer;
 use Interop\Queue\Topic;
+use InvalidArgumentException;
 use RdKafka\Conf;
 use RdKafka\KafkaConsumer;
 use RdKafka\Producer as VendorProducer;
@@ -54,8 +55,35 @@ public function __construct(array $config)
         $this->config = $config;
         $this->kafkaConsumers = [];
         $this->rdKafkaConsumers = [];
+        $this->configureSerializer($config);
+    }
+
+    /**
+     * @param array $config
+     * @return void
+     */
+    private function configureSerializer(array $config): void
+    {
+        if (!isset($config['serializer'])) {
+            $this->setSerializer(new JsonSerializer());
+            return;
+        }
 
-        $this->setSerializer(new JsonSerializer());
+        if (is_string($config['serializer'])) {
+            $this->setSerializer(new $config['serializer']());
+        } elseif (is_array($config['serializer']) && isset($config['serializer']['class'])) {
+            $serializerClass = $config['serializer']['class'];
+            $serializerOptions = $config['serializer']['options'] ?? [];
+            if (!empty($serializerOptions)) {
+                $this->setSerializer(new $serializerClass($serializerOptions));
+            } else {
+                $this->setSerializer(new $serializerClass());
+            }
+        } elseif ($config['serializer'] instanceof Serializer) {
+            $this->setSerializer($config['serializer']);
+        } else {
+            throw new InvalidArgumentException('Invalid serializer configuration');
+        }
     }
 
     /**

pkg/rdkafka/Tests/RdKafkaContextTest.php

@@ -8,6 +8,7 @@
 use Enqueue\RdKafka\Serializer;
 use Interop\Queue\Exception\InvalidDestinationException;
 use Interop\Queue\Exception\TemporaryQueueNotSupportedException;
+use InvalidArgumentException;
 use PHPUnit\Framework\TestCase;
 
 class RdKafkaContextTest extends TestCase
@@ -36,6 +37,27 @@ public function testShouldSetJsonSerializerInConstructor()
         $this->assertInstanceOf(JsonSerializer::class, $context->getSerializer());
     }
 
+    public function testShouldUseStringSerializerClassFromConfig()
+    {
+        $mockSerializerClass = get_class($this->createMock(Serializer::class));
+
+        $context = new RdKafkaContext([
+            'serializer' => $mockSerializerClass
+        ]);
+
+        $this->assertInstanceOf($mockSerializerClass, $context->getSerializer());
+    }
+
+    public function testShouldThrowExceptionOnInvalidSerializerConfig()
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('Invalid serializer configuration');
+
+        new RdKafkaContext([
+            'serializer' => 123
+        ]);
+    }
+
     public function testShouldAllowGetPreviouslySetSerializer()
     {
         $context = new RdKafkaContext([]);