Enforce DTOs for messaging

composer.json

@@ -7,13 +7,14 @@
     "php": "^8.2",
     "ext-json": "^8.2",
     "composer-plugin-api": "^2.0",
+    "eventsauce/object-hydrator": "*",
     "mammatus/app": "dev-master",
     "mammatus/kubernetes-attributes": "^1",
     "mammatus/kubernetes-contracts": "^1",
     "mammatus/kubernetes-events": "^1",
     "mammatus/life-cycle-events": "^2",
-    "mammatus/queue-attributes": "dev-main",
-    "mammatus/queue-contracts": "dev-main",
+    "mammatus/queue-attributes": "dev-enforce-DTOs-for-messaging",
+    "mammatus/queue-contracts": "dev-enforce-DTOs-for-messaging",
     "psr/container": "^1.1.2",
     "psr/event-dispatcher": "^1.0",
     "psr/log": "^2",

etc/generated_templates/AbstractList.php.twig

@@ -28,6 +28,8 @@ abstract class AbstractList
             '{{ worker.consumer.queue }}',
             {{ worker.consumer.concurrency }},
             \{{ worker.class }}::class,
+            '{{ worker.method }}',
+            \{{ worker.dtoClass }}::class,
             \json_decode('{{ worker.consumer.addOns|json_encode()|raw }}', true), /** @phpstan-ignore-line */
         );
 {% endfor %}

etc/qa/.phpunit.cache/test-results

@@ -1 +1 @@
-{"version":1,"defects":{"Mammatus\\Tests\\Cron\\AppTest::runAngry":8,"Mammatus\\Tests\\Cron\\AppTest::runNonAction":8,"Mammatus\\Tests\\Cron\\AppTest::runHappy":8,"Mammatus\\Tests\\Queue\\Composer\\InstallerTest::generate":7,"Mammatus\\Tests\\Queue\\Composer\\ItemTest::json":7,"Mammatus\\Tests\\Queue\\Kubernetes\\Helm\\CronJobsValuesTest::all":7,"Mammatus\\Tests\\Cron\\ManagerTest::runHappy":8,"Mammatus\\Tests\\Cron\\ManagerTest::runAngry":8,"Mammatus\\Tests\\Cron\\ManagerTest::notAnAction":8,"Mammatus\\Tests\\Queue\\AppTest::runHappy":8,"Mammatus\\Tests\\Queue\\AppTest::runAngry":8,"Mammatus\\Tests\\Queue\\AppTest::runNonAction":8,"Mammatus\\Tests\\Queue\\ManagerTest::runHappy":8,"Mammatus\\Tests\\Queue\\ManagerTest::runAngry":8,"Mammatus\\Tests\\Queue\\ManagerTest::notAnAction":8,"Mammatus\\Tests\\Queue\\ManagerTest::notAnWorker":8,"Mammatus\\Tests\\Queue\\AppTest::notAnWorker":7},"times":{"Mammatus\\Tests\\Cron\\AppTest::runHappy":0.076,"Mammatus\\Tests\\Cron\\AppTest::runAngry":0.002,"Mammatus\\Tests\\Cron\\AppTest::runNonAction":0.002,"Mammatus\\Tests\\Queue\\Composer\\InstallerTest::getSubscribedEvents":0.002,"Mammatus\\Tests\\Queue\\Composer\\InstallerTest::generate":0.982,"Mammatus\\Tests\\Queue\\Composer\\ItemTest::json":0.002,"Mammatus\\Tests\\Queue\\Kubernetes\\Helm\\CronJobsValuesTest::none":0.002,"Mammatus\\Tests\\Queue\\Kubernetes\\Helm\\CronJobsValuesTest::all":0.001,"Mammatus\\Tests\\Cron\\ManagerTest::runHappy":0.002,"Mammatus\\Tests\\Cron\\ManagerTest::runAngry":0.002,"Mammatus\\Tests\\Cron\\ManagerTest::notAnAction":0.002,"Mammatus\\Tests\\Queue\\AppTest::runHappy":0.053,"Mammatus\\Tests\\Queue\\AppTest::runAngry":0.003,"Mammatus\\Tests\\Queue\\AppTest::runNonAction":0.004,"Mammatus\\Tests\\Queue\\ManagerTest::runHappy":95.418,"Mammatus\\Tests\\Queue\\ManagerTest::runAngry":98.999,"Mammatus\\Tests\\Queue\\ManagerTest::notAnAction":99.056,"Mammatus\\Tests\\Queue\\ManagerTest::notAnWorker":0.004,"Mammatus\\Tests\\Queue\\AppTest::notAnWorker":0.003}}
+{"version":1,"defects":{"Mammatus\\Tests\\Cron\\AppTest::runAngry":8,"Mammatus\\Tests\\Cron\\AppTest::runNonAction":8,"Mammatus\\Tests\\Cron\\AppTest::runHappy":8,"Mammatus\\Tests\\Queue\\Composer\\InstallerTest::generate":7,"Mammatus\\Tests\\Queue\\Composer\\ItemTest::json":7,"Mammatus\\Tests\\Queue\\Kubernetes\\Helm\\CronJobsValuesTest::all":7,"Mammatus\\Tests\\Cron\\ManagerTest::runHappy":8,"Mammatus\\Tests\\Cron\\ManagerTest::runAngry":8,"Mammatus\\Tests\\Cron\\ManagerTest::notAnAction":8,"Mammatus\\Tests\\Queue\\AppTest::runHappy":8,"Mammatus\\Tests\\Queue\\AppTest::runAngry":8,"Mammatus\\Tests\\Queue\\AppTest::runNonAction":8,"Mammatus\\Tests\\Queue\\ManagerTest::runHappy":8,"Mammatus\\Tests\\Queue\\ManagerTest::runAngry":8,"Mammatus\\Tests\\Queue\\ManagerTest::notAnAction":8,"Mammatus\\Tests\\Queue\\ManagerTest::notAnWorker":8,"Mammatus\\Tests\\Queue\\AppTest::notAnWorker":7},"times":{"Mammatus\\Tests\\Cron\\AppTest::runHappy":0.076,"Mammatus\\Tests\\Cron\\AppTest::runAngry":0.002,"Mammatus\\Tests\\Cron\\AppTest::runNonAction":0.002,"Mammatus\\Tests\\Queue\\Composer\\InstallerTest::getSubscribedEvents":0.002,"Mammatus\\Tests\\Queue\\Composer\\InstallerTest::generate":1.178,"Mammatus\\Tests\\Queue\\Composer\\ItemTest::json":0.002,"Mammatus\\Tests\\Queue\\Kubernetes\\Helm\\CronJobsValuesTest::none":0.003,"Mammatus\\Tests\\Queue\\Kubernetes\\Helm\\CronJobsValuesTest::all":0.003,"Mammatus\\Tests\\Cron\\ManagerTest::runHappy":0.002,"Mammatus\\Tests\\Cron\\ManagerTest::runAngry":0.002,"Mammatus\\Tests\\Cron\\ManagerTest::notAnAction":0.002,"Mammatus\\Tests\\Queue\\AppTest::runHappy":0.061,"Mammatus\\Tests\\Queue\\AppTest::runAngry":0.003,"Mammatus\\Tests\\Queue\\AppTest::runNonAction":0.004,"Mammatus\\Tests\\Queue\\ManagerTest::runHappy":0.005,"Mammatus\\Tests\\Queue\\ManagerTest::runAngry":0.004,"Mammatus\\Tests\\Queue\\ManagerTest::notAnAction":99.056,"Mammatus\\Tests\\Queue\\ManagerTest::notAnWorker":0.004,"Mammatus\\Tests\\Queue\\AppTest::notAnWorker":0.003}}

etc/qa/phpcs.xml

@@ -9,5 +9,7 @@
     <file>../../src</file>
     <file>../../tests</file>
 
+    <exclude-pattern>src/Generated/*</exclude-pattern>
+
     <rule ref="WyriHaximus-OSS" />
 </ruleset>

etc/qa/phpstan.neon

@@ -1,6 +1,15 @@
 parameters:
+	excludePaths:
+		- ../../src/Generated/*
 	ignoreErrors:
 		- '#Method Mammatus\\Queue\\Consumer::__construct\(\) has a parameter \$container with a type declaration of Psr\\Container\\ContainerInterface, but containers should not be injected.#'
+		- '#thecodingmachine\/safe#'
+		-
+			message: '#file_put_contents blocks the event loop#'
+			path: ../../src/Composer/Plugin.php
+		-
+			message: '#Variable property access on Mammatus\\Queue\\Contracts\\Worker.#'
+			path: ../../src/Consumer.php
 	ergebnis:
 		noExtends:
 			classesAllowedToBeExtended:

src/BuildIn/EmptyMessage.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Mammatus\Queue\BuildIn;
+
+use Mammatus\Queue\Contracts\Work;
+
+final class EmptyMessage implements Work
+{
+}

src/BuildIn/Noop.php

@@ -4,17 +4,16 @@
 
 namespace Mammatus\Queue\BuildIn;
 
-use Interop\Queue\Message;
 use Mammatus\Queue\Attributes\Consumer;
 use Mammatus\Queue\Contracts\Worker;
 
 use function React\Async\await;
 use function WyriHaximus\React\timedPromise;
 
-#[Consumer(queue: 'noop', concurrency: 1)]
+#[Consumer(queue: 'noop', dtoClass: EmptyMessage::class, concurrency: 1)]
 final class Noop implements Worker
 {
-    public function perform(Message $message): void
+    public function perform(EmptyMessage $work): void
     {
         await(timedPromise(3, true));
     }

src/Composer/Collector.php

@@ -7,13 +7,17 @@
 use Mammatus\Kubernetes\Attributes\SplitOut;
 use Mammatus\Queue\Attributes\Consumer;
 use Roave\BetterReflection\Reflection\ReflectionClass;
+use Roave\BetterReflection\Reflection\ReflectionIntersectionType;
+use Roave\BetterReflection\Reflection\ReflectionUnionType;
 use WyriHaximus\Composer\GenerativePluginTooling\Item as ItemContract;
 use WyriHaximus\Composer\GenerativePluginTooling\ItemCollector;
 
 use function array_key_exists;
 
 final class Collector implements ItemCollector
 {
+    private const THE_NUMBER_OF_PARAMETERS_REQUIRED_FOR_A_METHOD_TO_BE_AN_EVENT_HANDLER_IS_ONE = 1;
+
     /** @return iterable<ItemContract> */
     public function collect(ReflectionClass $class): iterable
     {
@@ -27,11 +31,44 @@ public function collect(ReflectionClass $class): iterable
             return;
         }
 
-        /** @psalm-suppress ArgumentTypeCoercion */
-        yield new Item(
-            $class->getName(),
-            $attributes[Consumer::class], /** @phpstan-ignore-line */
-            array_key_exists(SplitOut::class, $attributes),
-        );
+        foreach ($class->getMethods() as $method) {
+            if (! $method->isPublic()) {
+                continue;
+            }
+
+            if ($method->isConstructor()) {
+                continue;
+            }
+
+            if ($method->isDestructor()) {
+                continue;
+            }
+
+            if ($method->getNumberOfParameters() !== self::THE_NUMBER_OF_PARAMETERS_REQUIRED_FOR_A_METHOD_TO_BE_AN_EVENT_HANDLER_IS_ONE) {
+                continue;
+            }
+
+            $messageDTOHolder = $method->getParameters()[0]->getType();
+            if ($messageDTOHolder instanceof ReflectionIntersectionType) {
+                continue;
+            }
+
+            if ($messageDTOHolder instanceof ReflectionUnionType) {
+                $messageDTOs = $messageDTOHolder->getTypes();
+            } else {
+                $messageDTOs = [$messageDTOHolder];
+            }
+
+            foreach ($messageDTOs as $messageDTO) {
+                /** @psalm-suppress ArgumentTypeCoercion */
+                yield new Item(
+                    $class->getName(),
+                    $method->getName(),
+                    (string) $messageDTO,
+                    $attributes[Consumer::class], /** @phpstan-ignore-line */
+                    array_key_exists(SplitOut::class, $attributes),
+                );
+            }
+        }
     }
 }

src/Composer/Item.php

@@ -13,6 +13,8 @@
     /** @param class-string $class */
     public function __construct(
         public string $class,
+        public string $method,
+        public string $dtoClass,
         public Consumer $consumer,
         public bool $splitOut,
     ) {
@@ -23,6 +25,8 @@ public function jsonSerialize(): array
     {
         return [
             'class' => $this->class,
+            'method' => $this->method,
+            'dtoClass' => $this->dtoClass,
             'consumer' => $this->consumer,
             'split_out' => $this->splitOut,
         ];

src/Composer/Plugin.php

@@ -4,6 +4,7 @@
 
 namespace Mammatus\Queue\Composer;
 
+use EventSauce\ObjectHydrator\ObjectMapperCodeGenerator;
 use Mammatus\Queue\Contracts\Worker;
 use WyriHaximus\Composer\GenerativePluginTooling\Filter\Class\ImplementsInterface;
 use WyriHaximus\Composer\GenerativePluginTooling\Filter\Class\IsInstantiable;
@@ -59,5 +60,18 @@ public function compile(string $rootPath, ItemContract ...$items): void
         $installPathList   = $rootPath . '/src/Generated/AbstractList.php';
         file_put_contents($installPathList, $classContentsList); /** @phpstan-ignore-line */
         chmod($installPathList, 0664); /** @phpstan-ignore-line */
+
+        $dtos = [];
+        foreach ($items as $item) {
+            if (! ($item instanceof Item)) {
+                continue;
+            }
+
+            $dtos[] = $item->dtoClass;
+        }
+
+        $hydratorGenerator = new ObjectMapperCodeGenerator();
+        $code              = $hydratorGenerator->dump($dtos, 'Mammatus\Queue\Generated\Hydrator');
+        file_put_contents($rootPath . '/src/Generated/Hydrator.php', $code);
     }
 }

src/Consumer.php

@@ -7,13 +7,16 @@
 use Interop\Queue as QueueInterop;
 use Mammatus\Queue\Contracts\Worker as WorkerContract;
 use Mammatus\Queue\Generated\AbstractList;
+use Mammatus\Queue\Generated\Hydrator;
 use Psr\Container\ContainerInterface;
 use Psr\Log\LoggerInterface;
 use React\Promise\PromiseInterface;
 use RuntimeException;
 use Throwable;
 use WyriHaximus\Broadcast\Contracts\Listener;
 
+use function is_array;
+use function json_decode;
 use function React\Async\async;
 use function React\Async\await;
 use function React\Promise\all;
@@ -26,6 +29,7 @@ final class Consumer extends AbstractList implements Listener
     public function __construct(
         private readonly ContainerInterface $container,
         private readonly QueueInterop\Context $context,
+        private readonly Hydrator $hydrator,
         private readonly LoggerInterface $logger,
     ) {
     }
@@ -65,7 +69,13 @@ private function consume(Worker $worker, WorkerContract $workerInstance): void
             }
 
             try {
-                $workerInstance->perform($message);
+                $json = json_decode($message->getBody(), true);
+                if (! is_array($json)) {
+                    throw new RuntimeException('Message is not valid JSON');
+                }
+
+                $dto = $this->hydrator->hydrateObject($worker->dtoClass, $json);
+                $workerInstance->$worker->method($dto);
                 $consumer->acknowledge($message);
             } catch (Throwable) { /** @phpstan-ignore-line */
                 $consumer->reject($message);

src/Worker.php

@@ -10,13 +10,16 @@
 {
     /**
      * @param class-string<WorkerContract> $class
+     * @param class-string                 $dtoClass
      * @param array<string, mixed>         $addOns
      */
     public function __construct(
         public string $type,
         public string $queue,
         public int $concurrency,
         public string $class,
+        public string $method,
+        public string $dtoClass,
         public array $addOns,
     ) {
     }

tests/Angry.php

@@ -14,7 +14,7 @@ public function __construct(private readonly Throwable $angry)
     {
     }
 
-    public function perform(Message $message): void
+    public function perform(Message $work): void
     {
         throw $this->angry;
     }

tests/Composer/ItemTest.php

@@ -6,6 +6,8 @@
 
 use Mammatus\Kubernetes\Attributes\Resources;
 use Mammatus\Queue\Attributes\Consumer;
+use Mammatus\Queue\BuildIn\EmptyMessage;
+use Mammatus\Queue\BuildIn\Noop;
 use Mammatus\Queue\Composer\Item;
 use WyriHaximus\TestUtilities\TestCase;
 
@@ -17,9 +19,12 @@ final class ItemTest extends TestCase
     public function json(): void
     {
         $item = new Item(
-            Item::class,
+            Noop::class,
+            'perform',
+            EmptyMessage::class,
             new Consumer(
                 'test',
+                EmptyMessage::class,
                 1337,
                 new Resources(
                     cpu: 0.666,
@@ -29,7 +34,7 @@ public function json(): void
             false,
         );
         self::assertSame(
-            '{"class":"Mammatus\\\\Queue\\\\Composer\\\\Item","consumer":{"addOns":[{"type":"container","helper":"mammatus.container.resources","arguments":{"cpu":"666m","memory":"3072Mi"}}],"queue":"test","concurrency":1337},"split_out":false}',
+            '{"class":"Mammatus\\\\Queue\\\\BuildIn\\\\Noop","method":"perform","dtoClass":"Mammatus\\\\Queue\\\\BuildIn\\\\EmptyMessage","consumer":{"addOns":[{"type":"container","helper":"mammatus.container.resources","arguments":{"cpu":"666m","memory":"3072Mi"}}],"queue":"test","dtoClass":"Mammatus\\\\Queue\\\\BuildIn\\\\EmptyMessage","concurrency":1337},"split_out":false}',
             json_encode($item),
         );
     }

tests/ConsumerFactory.php

@@ -7,6 +7,7 @@
 use Interop\Queue as QueueInterop;
 use Interop\Queue\Queue;
 use Mammatus\Queue\Consumer;
+use Mammatus\Queue\Generated\Hydrator;
 use Mockery;
 use Psr\Container\ContainerInterface;
 use Psr\Log\LoggerInterface;
@@ -29,7 +30,7 @@ public static function create(): array
             return true;
         })->between(0, PHP_INT_MAX)->andReturn($consumerInternal);
 
-        $consumer = new Consumer($container, $context, $logger);
+        $consumer = new Consumer($container, $context, new Hydrator(), $logger);
 
         return [
             $consumer,