Merge pull request php-enqueue#1114 from macghriogair/bugfix/queue-consume-catch-throwable

[fix] queue consumption: catch throwable for processing errors

Author: makasim

pkg/enqueue/Consumption/Context/ProcessorException.php

@@ -26,7 +26,7 @@ final class ProcessorException
     private $message;
 
     /**
-     * @var \Exception
+     * @var \Throwable
      */
     private $exception;
 
@@ -44,7 +44,7 @@ final class ProcessorException
      */
     private $logger;
 
-    public function __construct(Context $context, Consumer $consumer, Message $message, \Exception $exception, int $receivedAt, LoggerInterface $logger)
+    public function __construct(Context $context, Consumer $consumer, Message $message, \Throwable $exception, int $receivedAt, LoggerInterface $logger)
     {
         $this->context = $context;
         $this->consumer = $consumer;
@@ -69,7 +69,7 @@ public function getMessage(): Message
         return $this->message;
     }
 
-    public function getException(): \Exception
+    public function getException(): \Throwable
     {
         return $this->exception;
     }

pkg/enqueue/Consumption/QueueConsumer.php

@@ -195,7 +195,7 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
             if (null === $result) {
                 try {
                     $result = $processor->process($message, $this->interopContext);
-                } catch (\Exception $e) {
+                } catch (\Exception | \Throwable $e) {
                     $result = $this->onProcessorException($extension, $consumer, $message, $e, $receivedAt);
                 }
             }
@@ -303,7 +303,7 @@ private function onEnd(ExtensionInterface $extension, int $startTime, ?int $exit
      *
      * https://github.com/symfony/symfony/blob/cbe289517470eeea27162fd2d523eb29c95f775f/src/Symfony/Component/HttpKernel/EventListener/ExceptionListener.php#L77
      */
-    private function onProcessorException(ExtensionInterface $extension, Consumer $consumer, Message $message, \Exception $exception, int $receivedAt)
+    private function onProcessorException(ExtensionInterface $extension, Consumer $consumer, Message $message, \Throwable $exception, int $receivedAt)
     {
         $processorException = new ProcessorException($this->interopContext, $consumer, $message, $exception, $receivedAt, $this->logger);