Merge pull request #128 from cakephp/cake5-merge

merge master => cake5

Author: markstory

README.md

@@ -11,7 +11,7 @@ The plugin consists of a CakePHP shell wrapper and Queueing libraries for the [p
 
 ## Installation
 
-You can install this plugin into your CakePHP application using [Composer](http://getcomposer.org).
+You can install this plugin into your CakePHP application using [Composer](https://getcomposer.org).
 
 Run the following command
 ```sh

docs.Dockerfile

@@ -1,5 +1,5 @@
 # Generate the HTML output.
-FROM markstory/cakephp-docs-builder as builder
+FROM ghcr.io/cakephp/docs-builder as builder
 
 COPY docs /data/docs
 
@@ -8,7 +8,7 @@ RUN cd /data/docs-builder && \
   make website LANGS="en" SOURCE=/data/docs DEST=/data/website/
 
 # Build a small nginx container with just the static site in it.
-FROM markstory/cakephp-docs-builder:runtime as runtime
+FROM ghcr.io/cakephp/docs-builder:runtime as runtime
 
 # Configure search index script
 ENV LANGS="en"

docs/en/index.rst

@@ -118,7 +118,7 @@ A simple job that logs received messages would look like::
         use LogTrait;
 
         /**
-         * The maximum number of times the job may be attempted.
+         * The maximum number of times the job may be attempted. (optional property)
          * 
          * @var int|null
          */

src/Command/JobCommand.php

@@ -55,8 +55,11 @@ public function templateData(Arguments $arguments): array
     {
         $parentData = parent::templateData($arguments);
 
+        $maxAttempts = $arguments->getOption('max-attempts');
+
         $data = [
             'isUnique' => $arguments->getOption('unique'),
+            'maxAttempts' => $maxAttempts ? (int)$maxAttempts : null,
         ];
 
         return array_merge($parentData, $data);
@@ -77,6 +80,10 @@ public function buildOptionParser(ConsoleOptionParser $parser): ConsoleOptionPar
             ->addArgument('name', [
                 'help' => 'The name of the queue job class to create.',
             ])
+            ->addOption('max-attempts', [
+                'help' => 'The maximum number of times the job may be attempted.',
+                'default' => null,
+            ])
             ->addOption('unique', [
                 'help' => 'Whether there should be only one instance of a job on the queue at a time.',
                 'boolean' => true,

src/Command/WorkerCommand.php

@@ -129,10 +129,12 @@ protected function getQueueExtension(Arguments $args, LoggerInterface $logger):
 
         $limitAttempsExtension->getEventManager()->on(new FailedJobsListener());
 
+        $configKey = (string)$args->getOption('config');
+        $config = QueueManager::getConfig($configKey);
+
         $extensions = [
             new LoggerExtension($logger),
             $limitAttempsExtension,
-            new RemoveUniqueJobIdFromCacheExtension('Cake/Queue.queueUnique'),
         ];
 
         if (!is_null($args->getOption('max-jobs'))) {
@@ -145,6 +147,10 @@ protected function getQueueExtension(Arguments $args, LoggerInterface $logger):
             $extensions[] = new LimitConsumptionTimeExtension($endTime);
         }
 
+        if (isset($config['uniqueCacheKey'])) {
+            $extensions[] = new RemoveUniqueJobIdFromCacheExtension($config['uniqueCacheKey']);
+        }
+
         return new ChainExtension($extensions);
     }
 
@@ -171,16 +177,16 @@ protected function getLogger(Arguments $args): LoggerInterface
      */
     public function execute(Arguments $args, ConsoleIo $io): int
     {
-        $logger = $this->getLogger($args);
-        $processor = new Processor($logger, $this->container);
-        $extension = $this->getQueueExtension($args, $logger);
-
         $config = (string)$args->getOption('config');
         if (!Configure::check(sprintf('Queue.%s', $config))) {
             $io->error(sprintf('Configuration key "%s" was not found', $config));
             $this->abort();
         }
 
+        $logger = $this->getLogger($args);
+        $processor = new Processor($logger, $this->container);
+        $extension = $this->getQueueExtension($args, $logger);
+
         $hasListener = Configure::check(sprintf('Queue.%s.listener', $config));
         if ($hasListener) {
             $listenerClassName = Configure::read(sprintf('Queue.%s.listener', $config));
@@ -197,7 +203,7 @@ public function execute(Arguments $args, ConsoleIo $io): int
         $queue = $args->getOption('queue')
             ? (string)$args->getOption('queue')
             : Configure::read("Queue.{$config}.queue", 'default');
-        $processorName = $args->getOption('processor') ? (string)$args->getOption('processor') : null;
+        $processorName = $args->getOption('processor') ? (string)$args->getOption('processor') : 'default';
 
         $client->bindTopic($queue, $processor, $processorName);
         $client->consume($extension);

src/Consumption/LimitAttemptsExtension.php

@@ -64,13 +64,11 @@ public function onResult(MessageResult $context): void
         $attemptNumber = $message->getProperty(self::ATTEMPTS_PROPERTY, 0) + 1;
 
         if ($attemptNumber >= $maxAttempts) {
-            $originalResult = $context->getResult();
-
             $context->changeResult(
                 Result::reject(sprintf('The maximum number of %d allowed attempts was reached.', $maxAttempts))
             );
 
-            $exception = $originalResult instanceof Result ? $originalResult->getReason() : null;
+            $exception = (string)$message->getProperty('jobException');
 
             $this->dispatchEvent(
                 'Consumption.LimitAttemptsExtension.failed',

src/Queue/Processor.php

@@ -21,13 +21,13 @@
 use Cake\Queue\Job\Message;
 use Enqueue\Consumption\Result;
 use Error;
-use Exception;
 use Interop\Queue\Context;
 use Interop\Queue\Message as QueueMessage;
 use Interop\Queue\Processor as InteropProcessor;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use RuntimeException;
+use Throwable;
 
 class Processor implements InteropProcessor
 {
@@ -83,18 +83,20 @@ public function process(QueueMessage $message, Context $context): string|object
 
         try {
             $response = $this->processMessage($jobMessage);
-        } catch (Exception $e) {
+        } catch (Throwable $e) {
+            $message->setProperty('jobException', $e);
+
             $this->logger->debug(sprintf('Message encountered exception: %s', $e->getMessage()));
             $this->dispatchEvent('Processor.message.exception', [
                 'message' => $jobMessage,
                 'exception' => $e,
             ]);
 
-            return Result::requeue(sprintf('Exception occurred while processing message: %s', (string)$e));
+            return Result::requeue('Exception occurred while processing message');
         }
 
         if ($response === InteropProcessor::ACK) {
-            $this->logger->debug('Message processed sucessfully');
+            $this->logger->debug('Message processed successfully');
             $this->dispatchEvent('Processor.message.success', ['message' => $jobMessage]);
 
             return InteropProcessor::ACK;

src/QueueManager.php

@@ -89,10 +89,11 @@ public static function setConfig(string|array $key, ?array $config = null): void
             }
 
             return;
+        } elseif (is_array($key)) {
+            throw new LogicException('If config is not null, key must be a string.');
         }
 
         if (isset(static::$_config[$key])) {
-            /** @psalm-suppress PossiblyInvalidArgument */
             throw new BadMethodCallException(sprintf('Cannot reconfigure existing key `%s`', $key));
         }
 
@@ -127,10 +128,11 @@ public static function setConfig(string|array $key, ?array $config = null): void
 
             $cacheConfig = array_merge($cacheDefaults, $config['uniqueCache']);
 
-            Cache::setConfig('Cake/Queue.queueUnique', $cacheConfig);
+            $config['uniqueCacheKey'] = "Cake/Queue.queueUnique.{$key}";
+
+            Cache::setConfig($config['uniqueCacheKey'], $cacheConfig);
         }
 
-        /** @psalm-suppress InvalidPropertyAssignmentValue */
         static::$_config[$key] = $config;
     }
 
@@ -228,15 +230,15 @@ public static function push(string|array $className, array $data = [], array $op
 
         /** @psalm-suppress InvalidPropertyFetch */
         if (!empty($class::$shouldBeUnique)) {
-            if (!Cache::getConfig('Cake/Queue.queueUnique')) {
+            if (empty($config['uniqueCache'])) {
                 throw new InvalidArgumentException(
                     "$class::\$shouldBeUnique is set to `true` but `uniqueCache` configuration is missing."
                 );
             }
 
             $uniqueId = static::getUniqueId($class, $method, $data);
 
-            if (Cache::read($uniqueId, 'Cake/Queue.queueUnique')) {
+            if (Cache::read($uniqueId, $config['uniqueCacheKey'])) {
                 if ($logger) {
                     $logger->debug(
                         "An identical instance of $class already exists on the queue. This push will be ignored."
@@ -279,7 +281,7 @@ public static function push(string|array $className, array $data = [], array $op
         if (!empty($class::$shouldBeUnique)) {
             $uniqueId = static::getUniqueId($class, $method, $data);
 
-            Cache::add($uniqueId, true, 'Cake/Queue.queueUnique');
+            Cache::add($uniqueId, true, $config['uniqueCacheKey']);
         }
     }
 

templates/bake/job.twig

@@ -27,13 +27,15 @@ use Interop\Queue\Processor;
  */
 class {{ name }}Job implements JobInterface
 {
+{% if maxAttempts %}
     /**
      * The maximum number of times the job may be attempted.
      * 
      * @var int|null
      */
-    public static $maxAttempts = 3;
+    public static $maxAttempts = {{ maxAttempts }};
 
+{% endif %}
 {% if isUnique %}
     /**
      * Whether there should be only one instance of a job on the queue at a time. (optional property)

tests/TestCase/Command/WorkerCommandTest.php

@@ -293,4 +293,31 @@ public function testQueueProcessesJobWithDIService()
 
         $this->assertDebugLogContains('Debug job was run with service infotext');
     }
+
+    /**
+     * Test that queue will process when a unique cache is configured.
+     *
+     * @runInSeparateProcess
+     */
+    public function testQueueProcessesWithUniqueCacheConfigured()
+    {
+        $config = [
+            'queue' => 'default',
+            'url' => 'file:///' . TMP . DS . 'queue',
+            'receiveTimeout' => 100,
+            'uniqueCache' => [
+                'engine' => 'File',
+            ],
+        ];
+        Configure::write('Queue', ['default' => $config]);
+
+        Log::setConfig('debug', [
+            'className' => 'Array',
+            'levels' => ['notice', 'info', 'debug'],
+        ]);
+
+        $this->exec('queue worker --max-jobs=1 --logger=debug --verbose');
+
+        $this->assertDebugLogContains('Consumption has started');
+    }
 }