Add fallback to JsonSerializer in configureSerializer

- Ensures robustness by providing a default serializer
- Refactoring

Author: lenchester

pkg/rdkafka/RdKafkaContext.php

@@ -16,7 +16,6 @@
 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;
@@ -58,34 +57,6 @@ public function __construct(array $config)
         $this->configureSerializer($config);
     }
 
-    /**
-     * @param array $config
-     * @return void
-     */
-    private function configureSerializer(array $config): void
-    {
-        if (!isset($config['serializer'])) {
-            $this->setSerializer(new JsonSerializer());
-            return;
-        }
-
-        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');
-        }
-    }
-
     /**
      * @return RdKafkaMessage
      */
@@ -208,6 +179,58 @@ public static function getLibrdKafkaVersion(): string
         return "$major.$minor.$patch";
     }
 
+    /**
+     * @return void
+     *              JsonSerializer should be the default fallback if no serializer is specified
+     */
+    private function configureSerializer(array $config): void
+    {
+        if (!isset($config['serializer'])) {
+            $this->setSerializer(new JsonSerializer());
+
+            return;
+        }
+
+        $serializer = $config['serializer'];
+
+        if ($serializer instanceof Serializer) {
+            $this->setSerializer($serializer);
+
+            return;
+        }
+
+        $serializerClass = $this->resolveSerializerClass($serializer);
+
+        if (!class_exists($serializerClass) || !is_a($serializerClass, Serializer::class, true)) {
+            throw $this->createInvalidSerializerException($serializerClass);
+        }
+
+        $serializerOptions = $serializer['options'] ?? [];
+        $this->setSerializer(new $serializerClass($serializerOptions));
+    }
+
+    private function resolveSerializerClass(mixed $serializer): string
+    {
+        if (is_string($serializer)) {
+            return $serializer;
+        }
+
+        if (is_array($serializer) && isset($serializer['class'])) {
+            return $serializer['class'];
+        }
+
+        throw $this->createInvalidSerializerException($serializer);
+    }
+
+    private function createInvalidSerializerException(mixed $value): \InvalidArgumentException
+    {
+        return new \InvalidArgumentException(sprintf(
+            'Invalid serializer configuration. Expected "serializer" to be a string, an array with a "class" key, or a %s instance. Received %s instead.',
+            Serializer::class,
+            get_debug_type($value)
+        ));
+    }
+
     private function getConf(): Conf
     {
         if (null === $this->conf) {

pkg/rdkafka/Tests/RdKafkaContextTest.php

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