Add DTO (Data Transfer Object) support to Queue plugin

composer.json

@@ -21,10 +21,11 @@
         "source": "https://github.com/cakephp/queue"
     },
     "require": {
-        "php": ">=8.1",
-        "cakephp/cakephp": "^5.1.0",
+        "php": ">=8.2",
+        "cakephp/cakephp": "^5.4.0",
         "enqueue/simple-client": "^0.10",
-        "psr/log": "^3.0"
+        "psr/log": "^3.0",
+        "ramsey/uuid": "^4.7.0"
     },
     "require-dev": {
         "cakephp/bake": "^3.5.1",

docs/en/jobs.md

@@ -60,7 +60,7 @@ Returning any other value is treated as a failure and results in the message bei
 ## Job Properties
 
 - `maxAttempts` limits how many times a job can be retried after an exception or explicit `Processor::REQUEUE`. If unset, the worker's `--max-attempts` option applies. If neither is set, retries are unlimited.
-- `shouldBeUnique` allows only one queued copy of the same job class, method, and payload. Duplicate pushes are ignored. This requires `uniqueCache` in the queue configuration.
+- `shouldBeUnique` allows only one queued copy of the same job class, method, and payload. Duplicate pushes are ignored. This requires `uniqueCache` in the queue configuration. When the payload is a DTO, its class is also factored into the uniqueness check, so two different DTO classes with coincidentally identical data are never treated as duplicates of each other.
 
 ## Queueing Jobs
 
@@ -89,3 +89,69 @@ Supported options:
 - `expires`: expire the message after a number of seconds if it has not been consumed.
 - `priority`: one of `\Enqueue\Client\MessagePriority::VERY_LOW`, `LOW`, `NORMAL`, `HIGH`, or `VERY_HIGH`.
 - `queue`: queue name to use. Defaults to the configured queue, then `default`.
+
+## Dispatching and Receiving DTOs
+
+Instead of an array, `QueueManager::push()` also accepts a DTO object as the payload:
+
+```php
+use App\Dto\OrderDto;
+use App\Job\ProcessOrderJob;
+use Cake\Queue\QueueManager;
+
+$order = new OrderDto(id: 7, customer: 'Acme Corp');
+
+QueueManager::push(ProcessOrderJob::class, $order);
+```
+
+The DTO is serialized into the same JSON-safe array that a plain array payload would produce (via `jsonSerialize()` when the DTO implements `JsonSerializable`, otherwise its public properties). The DTO's class name is also recorded on the message as metadata (used for `shouldBeUnique` hashing and debugging). If you only have an array at the dispatch site but still want that metadata recorded, pass the class via the `dtoClass` option:
+
+```php
+QueueManager::push(ProcessOrderJob::class, $data, [
+    'dtoClass' => OrderDto::class,
+]);
+```
+
+A plain array push with no `dtoClass` option behaves exactly as before; the message body is unchanged.
+
+### Receiving a DTO in a job
+
+Call `Message::getDto()` with the class your job expects. The expected type comes from your code, not from the message body — that way a tampered queue message cannot choose which class gets instantiated. `getArgument()` keeps returning the raw array:
+
+```php
+public function execute(Message $message): ?string
+{
+    $order = $message->getDto(OrderDto::class);
+    $id = $message->getArgument('id'); // the raw array is still available
+
+    return Processor::ACK;
+}
+```
+
+If the payload cannot be hydrated into the given class, `getDto()` throws. Jobs that still need to accept legacy array-only messages can catch that exception (or keep using `getArgument()` only) while they migrate.
+
+### Supported DTO classes
+
+Hydration mirrors the DTO conventions used elsewhere in CakePHP (`#[RequestToDto]` for controllers, `SelectQuery::projectAs()` for the ORM), so the same DTO class can be reused across all three:
+
+- **Constructor reflection** — a plain class (typically `readonly`) with typed, named constructor parameters. Nested DTOs are resolved from the parameter's type hint, and arrays of DTOs via the `#[CollectionOf]` attribute:
+
+  ```php
+  use Cake\ORM\Attribute\CollectionOf;
+
+  readonly class OrderDto
+  {
+      /**
+       * @param array<int, \App\Dto\OrderItemDto> $items
+       */
+      public function __construct(
+          public int $id,
+          public string $customer,
+          #[CollectionOf(OrderItemDto::class)]
+          public array $items = [],
+      ) {
+      }
+  }
+  ```
+
+- **`createFromArray()` factory** — if the DTO class defines a static `createFromArray(array $data, bool $nested = false): static` method, it's used instead of reflection.

src/Consumption/LimitAttemptsExtension.php

@@ -12,9 +12,6 @@
 
 class LimitAttemptsExtension implements MessageResultExtensionInterface
 {
-    /**
-     * @use \Cake\Event\EventDispatcherTrait<\Cake\Queue\Job\Message>
-     */
     use EventDispatcherTrait;
 
     /**

src/Consumption/RemoveUniqueJobIdFromCacheExtension.php

@@ -37,7 +37,7 @@ public function onResult(MessageResult $context): void
 
         $data = $jobMessage->getArgument();
 
-        $uniqueId = QueueManager::getUniqueId($class, $method, $data);
+        $uniqueId = QueueManager::getUniqueId($class, $method, $data, $jobMessage->getDtoClass());
 
         Cache::delete($uniqueId, $this->cache);
     }

src/Dto/DtoManager.php

@@ -0,0 +1,106 @@
+<?php
+declare(strict_types=1);
+
+/**
+ * CakePHP(tm) : Rapid Development Framework (https://cakephp.org)
+ * Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
+ *
+ * Licensed under The MIT License
+ * For full copyright and license information, please see the LICENSE.txt
+ * Redistributions of files must retain the above copyright notice.
+ *
+ * @copyright     Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
+ * @link          https://cakephp.org CakePHP(tm) Project
+ * @since         3.0.0
+ * @license       https://opensource.org/licenses/MIT MIT License
+ */
+namespace Cake\Queue\Dto;
+
+use Cake\ORM\ResultSetFactory;
+use InvalidArgumentException;
+use JsonSerializable;
+
+/**
+ * Serializes DTO objects for queue transport and hydrates them back on the
+ * receiving side.
+ *
+ * Hydration reuses the CakePHP 5.4 DTO support via `ResultSetFactory::getDtoHydrator()`,
+ * which handles both a static `createFromArray($data, $nested)` factory method
+ * (cakephp-dto style) and plain DTOs mapped through `Cake\ORM\DtoMapper` (constructor
+ * parameters, nested DTO type-hints and the `#[CollectionOf]` attribute).
+ */
+class DtoManager
+{
+    /**
+     * Serialize a DTO (or array) into an array suitable for queue transport.
+     *
+     * Nested objects are converted to arrays so the resulting data only contains
+     * JSON-encodable scalars.
+     *
+     * @param array<string, mixed>|object $data Data or DTO object to serialize.
+     * @return array<string, mixed> Serialized data.
+     */
+    public static function serialize(array|object $data): array
+    {
+        if ($data instanceof JsonSerializable) {
+            $data = $data->jsonSerialize();
+        }
+
+        if (is_object($data)) {
+            $data = get_object_vars($data);
+        }
+
+        if (!is_array($data)) {
+            throw new InvalidArgumentException(
+                'DTO data could not be serialized into an array. `jsonSerialize()` must return an array or object.',
+            );
+        }
+
+        return self::toScalarArray($data);
+    }
+
+    /**
+     * Hydrate queue data back into a DTO instance.
+     *
+     * This method is not safe to use with user-defined `dtoClass` values.
+     * @template T of object
+     * @param array<string, mixed> $data Serialized data.
+     * @param class-string<T> $dtoClass DTO class name.
+     * @return T Hydrated DTO instance.
+     * @throws \InvalidArgumentException When the DTO class does not exist.
+     */
+    public static function deserialize(array $data, string $dtoClass): object
+    {
+        if (!class_exists($dtoClass)) {
+            throw new InvalidArgumentException(sprintf('DTO class `%s` does not exist.', $dtoClass));
+        }
+
+        $dto = (new ResultSetFactory())->hydrateDto($data, $dtoClass);
+        assert($dto instanceof $dtoClass);
+
+        return $dto;
+    }
+
+    /**
+     * Recursively convert any nested objects into arrays.
+     *
+     * @param array<string, mixed> $data The data to convert.
+     * @return array<string, mixed> The converted data.
+     */
+    protected static function toScalarArray(array $data): array
+    {
+        foreach ($data as $key => $value) {
+            if (is_object($value)) {
+                $value = $value instanceof JsonSerializable
+                    ? $value->jsonSerialize()
+                    : get_object_vars($value);
+            }
+
+            if (is_array($value)) {
+                $data[$key] = self::toScalarArray($value);
+            }
+        }
+
+        return $data;
+    }
+}

src/Job/JobInterface.php

@@ -22,7 +22,6 @@ interface JobInterface
      * Executes logic for Job
      *
      * @param \Cake\Queue\Job\Message $message job message
-     * @return string|null
      */
     public function execute(Message $message): ?string;
 }

src/Job/MailerJob.php

@@ -29,7 +29,6 @@ class MailerJob implements JobInterface
      * Constructs and dispatches the event from a job message
      *
      * @param \Cake\Queue\Job\Message $message job message
-     * @return string|null
      */
     public function execute(Message $message): ?string
     {

src/Job/Message.php

@@ -17,10 +17,12 @@
 namespace Cake\Queue\Job;
 
 use Cake\Core\ContainerInterface;
+use Cake\Queue\Dto\DtoManager;
 use Cake\Utility\Hash;
 use Closure;
 use Interop\Queue\Context;
 use Interop\Queue\Message as QueueMessage;
+use InvalidArgumentException;
 use JsonSerializable;
 use RuntimeException;
 
@@ -33,6 +35,13 @@ class Message implements JsonSerializable
 
     protected ?Closure $callable = null;
 
+    protected ?object $dto = null;
+
+    /**
+     * @var class-string|null
+     */
+    protected ?string $dtoHydratedAs = null;
+
     /**
      * @param \Interop\Queue\Message $originalMessage Queue message.
      * @param \Interop\Queue\Context $context Context.
@@ -126,7 +135,6 @@ public function getTarget(): array
     /**
      * @param mixed $key Key
      * @param mixed $default Default value.
-     * @return mixed
      */
     public function getArgument(mixed $key = null, mixed $default = null): mixed
     {
@@ -146,9 +154,58 @@ public function getArgument(mixed $key = null, mixed $default = null): mixed
     }
 
     /**
-     * The maximum number of attempts allowed by the job.
+     * Get the DTO class name recorded on the message body at dispatch time, if any.
+     *
+     * This value is metadata for uniqueness hashing and debugging. It is never used
+     * as the hydration target — pass the expected class to `getDto()` instead.
+     *
+     * @return class-string|null
+     */
+    public function getDtoClass(): ?string
+    {
+        $dtoClass = $this->parsedBody['dtoClass'] ?? null;
+        if (!is_string($dtoClass) || !class_exists($dtoClass)) {
+            return null;
+        }
+
+        return $dtoClass;
+    }
+
+    /**
+     * Hydrate the message data into the expected DTO class.
      *
-     * @return int|null
+     * The class name must come from application code, not from the message body.
+     * That keeps queue consumers safe if a message is tampered with: only the type
+     * the job asks for is ever instantiated.
+     *
+     * @template T of object
+     * @param class-string<T> $dtoClass The DTO class the job expects.
+     * @return T
+     * @throws \InvalidArgumentException When `$dtoClass` does not exist or cannot be hydrated.
+     */
+    public function getDto(string $dtoClass): object
+    {
+        if ($this->dto !== null && $this->dtoHydratedAs === $dtoClass) {
+            assert($this->dto instanceof $dtoClass);
+
+            return $this->dto;
+        }
+
+        if (!class_exists($dtoClass)) {
+            throw new InvalidArgumentException(sprintf('DTO class `%s` does not exist.', $dtoClass));
+        }
+
+        $dto = DtoManager::deserialize($this->getArgument(), $dtoClass);
+        assert($dto instanceof $dtoClass);
+
+        $this->dto = $dto;
+        $this->dtoHydratedAs = $dtoClass;
+
+        return $dto;
+    }
+
+    /**
+     * The maximum number of attempts allowed by the job.
      */
     public function getMaxAttempts(): ?int
     {

src/Queue/Processor.php

@@ -31,9 +31,6 @@
 
 class Processor implements InteropProcessor
 {
-    /**
-     * @use \Cake\Event\EventDispatcherTrait<\Cake\Queue\Queue\Processor>
-     */
     use EventDispatcherTrait;
 
     /**