[Kafka Messenger] Added GeneratorMap support

Exception/DiscriminatorMapException.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace FRZB\Component\Messenger\Bridge\Kafka\Exception;
+
+use Fp\Collections\Entry;
+use Fp\Collections\HashMap;
+
+final class DiscriminatorMapException extends \LogicException
+{
+    private const MISSING_PARAMETER_ERROR_MESSAGE = 'Missing property "%s" with class "%s"';
+    private const MISSING_TYPE_PROPERTY_ERROR_MESSAGE = 'Missing mapping for type "%s" in class "%s" with mapping "%s"';
+
+    public static function fromThrowable(\Throwable $previous): self
+    {
+        return new self($previous->getMessage(), $previous->getCode(), $previous);
+    }
+
+    public static function missingProperty(string $className, string $propertyName, ?\Throwable $previous = null): self
+    {
+        $message = sprintf(self::MISSING_PARAMETER_ERROR_MESSAGE, $propertyName, $className);
+
+        return new self($message, previous: $previous);
+    }
+
+    public static function missingTypeProperty(string $className, string $typeProperty, array $mapping, ?\Throwable $previous = null): self
+    {
+        $map = HashMap::collect($mapping)
+            ->map(static fn (Entry $entry) => "{$entry->key}: {$entry->value}")
+            ->toArrayList()
+            ->mkString()
+        ;
+
+        $message = sprintf(self::MISSING_TYPE_PROPERTY_ERROR_MESSAGE, $typeProperty, $className, $map);
+
+        return new self($message, previous: $previous);
+    }
+}

Helper/ClassHelper.php

@@ -0,0 +1,68 @@
+<?php
+
+declare(strict_types=1);
+
+namespace FRZB\Component\Messenger\Bridge\Kafka\Helper;
+
+use Fp\Collections\ArrayList;
+use FRZB\Component\Messenger\Bridge\Kafka\Exception\DiscriminatorMapException;
+use JetBrains\PhpStorm\Immutable;
+use Symfony\Component\Serializer\Annotation\DiscriminatorMap;
+
+#[Immutable]
+final class ClassHelper
+{
+    public static function getClassName(string $targetClass, array|string $targetPayload): string
+    {
+        try {
+            $targetPayload = \is_array($targetPayload) ? $targetPayload : json_decode($targetPayload, true, flags: \JSON_THROW_ON_ERROR);
+        } catch (\JsonException) {
+            $targetPayload = [];
+        }
+
+        if ($discriminatorMap = self::getAttribute($targetClass, DiscriminatorMap::class)) {
+            $property = $discriminatorMap->getTypeProperty();
+            $mapping = $discriminatorMap->getMapping();
+            $parameter = $targetPayload[$property] ?? throw DiscriminatorMapException::missingProperty($targetClass, $property);
+            $targetClass = $mapping[$parameter] ?? throw DiscriminatorMapException::missingTypeProperty($targetClass, $property, $mapping);
+        }
+
+        return $targetClass;
+    }
+
+    /**
+     * @template T
+     *
+     * @param class-string<T> $attributeClass
+     *
+     * @return null|T
+     */
+    public static function getAttribute(string|object $target, string $attributeClass): ?object
+    {
+        return ArrayList::collect(self::getAttributes($target, $attributeClass))
+            ->firstElement()
+            ->get()
+        ;
+    }
+
+    /**
+     * @template T
+     *
+     * @param class-string<T> $attributeClass
+     *
+     * @return array<T>
+     */
+    public static function getAttributes(string|object $target, string $attributeClass): array
+    {
+        try {
+            $attributes = (new \ReflectionClass($target))->getAttributes($attributeClass);
+        } catch (\ReflectionException) {
+            $attributes = [];
+        }
+
+        return ArrayList::collect($attributes)
+            ->map(static fn (\ReflectionAttribute $a) => $a->newInstance())
+            ->toArray()
+        ;
+    }
+}

Integration/MessageSerializer.php

@@ -4,6 +4,7 @@
 
 namespace FRZB\Component\Messenger\Bridge\Kafka\Integration;
 
+use FRZB\Component\Messenger\Bridge\Kafka\Helper\ClassHelper;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Transport\Serialization\Serializer;
 
@@ -26,6 +27,8 @@ private static function getBody(array $decodedEnvelope): ?string
 
     private static function getHeaders(array $decodedEnvelope): array
     {
-        return array_merge($decodedEnvelope['headers'] ?? [], ['type' => static::getMessageType()]);
+        $className = ClassHelper::getClassName(static::getMessageType(), self::getBody($decodedEnvelope));
+
+        return [...$decodedEnvelope['headers'] ?? [], ...['type' => $className]];
     }
 }