port over enqueue/simple-client

Author: LordSimal

composer.json

@@ -22,9 +22,12 @@
     },
     "require": {
         "php": ">=8.1",
-        "cakephp/cakephp": "dev-5.next as 5.1.0",
-        "enqueue/simple-client": "^0.10",
-        "psr/log": "^3.0"
+        "cakephp/cakephp": "^5.1.0",
+        "psr/log": "^3.0",
+        "enqueue/enqueue": "^0.10",
+        "queue-interop/amqp-interop": "^0.8.2",
+        "queue-interop/queue-interop": "^0.8",
+        "symfony/config": "^5.4|^6.0|^7.0"
     },
     "require-dev": {
         "cakephp/bake": "dev-3.next",

src/Command/RequeueCommand.php

@@ -138,7 +138,7 @@ public function execute(Arguments $args, ConsoleIo $io): void
                         'config' => $failedJob->config,
                         'priority' => $failedJob->priority,
                         'queue' => $failedJob->queue,
-                    ]
+                    ],
                 );
 
                 $failedJobsTable->deleteOrFail($failedJob);

src/Command/WorkerCommand.php

@@ -110,7 +110,7 @@ public function getOptionParser(): ConsoleOptionParser
             'short' => 'a',
         ]);
         $parser->setDescription(
-            'Runs a queue worker that consumes from the named queue.'
+            'Runs a queue worker that consumes from the named queue.',
         );
 
         return $parser;

src/Consumption/LimitAttemptsExtension.php

@@ -65,15 +65,15 @@ public function onResult(MessageResult $context): void
 
         if ($attemptNumber >= $maxAttempts) {
             $context->changeResult(
-                Result::reject(sprintf('The maximum number of %d allowed attempts was reached.', $maxAttempts))
+                Result::reject(sprintf('The maximum number of %d allowed attempts was reached.', $maxAttempts)),
             );
 
             $exception = (string)$message->getProperty('jobException');
 
             $this->dispatchEvent(
                 'Consumption.LimitAttemptsExtension.failed',
                 ['exception' => $exception, 'logger' => $context->getLogger()],
-                $jobMessage
+                $jobMessage,
             );
 
             return;
@@ -88,7 +88,7 @@ public function onResult(MessageResult $context): void
         $producer->send($consumer->getQueue(), $newMessage);
 
         $context->changeResult(
-            Result::reject('A copy of the message was sent with an incremented attempt count.')
+            Result::reject('A copy of the message was sent with an incremented attempt count.'),
         );
     }
 }

src/Consumption/LimitConsumedMessagesExtension.php

@@ -79,7 +79,7 @@ protected function shouldBeStopped(LoggerInterface $logger): bool
             $logger->debug(sprintf(
                 '[LimitConsumedMessagesExtension] Message consumption is interrupted since the message limit ' .
                 'reached. limit: "%s"',
-                $this->messageLimit
+                $this->messageLimit,
             ));
 
             return true;