symfony · fabpot · Jul 3, 2021 · Dec 23, 2020
diff --git a/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php b/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php
@@ -20,6 +20,7 @@
 use Symfony\Component\Messenger\EventListener\DispatchPcntlSignalListener;
 use Symfony\Component\Messenger\EventListener\SendFailedMessageForRetryListener;
 use Symfony\Component\Messenger\EventListener\SendFailedMessageToFailureTransportListener;
+use Symfony\Component\Messenger\EventListener\StopWorkerOnCustomStopExceptionListener;
 use Symfony\Component\Messenger\EventListener\StopWorkerOnRestartSignalListener;
 use Symfony\Component\Messenger\EventListener\StopWorkerOnSigtermSignalListener;
 use Symfony\Component\Messenger\Middleware\AddBusNameStampMiddleware;
@@ -193,6 +194,9 @@
         ->set('messenger.listener.stop_worker_on_sigterm_signal_listener', StopWorkerOnSigtermSignalListener::class)
             ->tag('kernel.event_subscriber')
 
+        ->set('messenger.listener.stop_worker_on_stop_exception_listener', StopWorkerOnCustomStopExceptionListener::class)
+            ->tag('kernel.event_subscriber')
+
         ->set('messenger.routable_message_bus', RoutableMessageBus::class)
             ->args([
                 abstract_arg('message bus locator'),

diff --git a/src/Symfony/Bundle/FrameworkBundle/composer.json b/src/Symfony/Bundle/FrameworkBundle/composer.json
@@ -48,7 +48,7 @@
         "symfony/http-client": "^4.4|^5.0|^6.0",
         "symfony/lock": "^4.4|^5.0|^6.0",
         "symfony/mailer": "^5.2|^6.0",
-        "symfony/messenger": "^5.2|^6.0",
+        "symfony/messenger": "^5.4|^6.0",
         "symfony/mime": "^4.4|^5.0|^6.0",
         "symfony/notifier": "^5.3|^6.0",
         "symfony/allmysms-notifier": "^5.3|^6.0",
@@ -115,7 +115,7 @@
         "symfony/form": "<5.2",
         "symfony/lock": "<4.4",
         "symfony/mailer": "<5.2",
-        "symfony/messenger": "<4.4",
+        "symfony/messenger": "<5.4",
         "symfony/mime": "<4.4",
         "symfony/property-info": "<4.4",
         "symfony/property-access": "<5.3",

diff --git a/src/Symfony/Component/Messenger/CHANGELOG.md b/src/Symfony/Component/Messenger/CHANGELOG.md
@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+5.4
+---
+
+ * Add `StopWorkerExceptionInterface` and its implementation `StopWorkerException` to stop the worker.
+
 5.3
 ---
 

diff --git a/src/Symfony/Component/Messenger/EventListener/StopWorkerOnCustomStopExceptionListener.php b/src/Symfony/Component/Messenger/EventListener/StopWorkerOnCustomStopExceptionListener.php
@@ -0,0 +1,57 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\EventListener;
+
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use Symfony\Component\Messenger\Event\WorkerRunningEvent;
+use Symfony\Component\Messenger\Exception\HandlerFailedException;
+use Symfony\Component\Messenger\Exception\StopWorkerExceptionInterface;
+
+/**
+ * @author Grégoire Pineau <lyrixx@lyrixx.info>
+ */
+class StopWorkerOnCustomStopExceptionListener implements EventSubscriberInterface
+{
+    private $stop = false;
+
+    public function onMessageFailed(WorkerMessageFailedEvent $event): void
+    {
+        $th = $event->getThrowable();
+        if ($th instanceof StopWorkerExceptionInterface) {
+            $this->stop = true;
+        }
+        if ($th instanceof HandlerFailedException) {
+            foreach ($th->getNestedExceptions() as $e) {
+                if ($e instanceof StopWorkerExceptionInterface) {
+                    $this->stop = true;
+                    break;
+                }
+            }
+        }
+    }
+
+    public function onWorkerRunning(WorkerRunningEvent $event): void
+    {
+        if ($this->stop) {
+            $event->getWorker()->stop();
+        }
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            WorkerMessageFailedEvent::class => 'onMessageFailed',
+            WorkerRunningEvent::class => 'onWorkerRunning',
+        ];
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Exception/StopWorkerException.php b/src/Symfony/Component/Messenger/Exception/StopWorkerException.php
@@ -0,0 +1,23 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Exception;
+
+/**
+ * @author Grégoire Pineau <lyrixx@lyrixx.info>
+ */
+class StopWorkerException extends RuntimeException implements StopWorkerExceptionInterface
+{
+    public function __construct(string $message = 'Worker should stop.', ?\Throwable $previous = null)
+    {
+        parent::__construct($message, 0, $previous);
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Exception/StopWorkerExceptionInterface.php b/src/Symfony/Component/Messenger/Exception/StopWorkerExceptionInterface.php
@@ -0,0 +1,19 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Exception;
+
+/**
+ * @author Grégoire Pineau <lyrixx@lyrixx.info>
+ */
+interface StopWorkerExceptionInterface extends \Throwable
+{
+}
diff --git a/...y/Component/Messenger/Tests/EventListener/StopWorkerOnCustomStopExceptionListenerTest.php b/...y/Component/Messenger/Tests/EventListener/StopWorkerOnCustomStopExceptionListenerTest.php
@@ -0,0 +1,57 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Tests\EventListener;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use Symfony\Component\Messenger\Event\WorkerRunningEvent;
+use Symfony\Component\Messenger\EventListener\StopWorkerOnCustomStopExceptionListener;
+use Symfony\Component\Messenger\Exception\HandlerFailedException;
+use Symfony\Component\Messenger\Exception\StopWorkerException;
+use Symfony\Component\Messenger\Exception\StopWorkerExceptionInterface;
+use Symfony\Component\Messenger\Worker;
+
+class StopWorkerOnCustomStopExceptionListenerTest extends TestCase
+{
+    public function provideTests(): \Generator
+    {
+        yield 'it should not stop (1)' => [new \Exception(), false];
+        yield 'it should not stop (2)' => [new HandlerFailedException(new Envelope(new \stdClass()), [new \Exception()]), false];
+
+        $t = new class() extends \Exception implements StopWorkerExceptionInterface {};
+        yield 'it should stop with custom exception' => [$t, true];
+        yield 'it should stop with core exception' => [new StopWorkerException(), true];
+
+        yield 'it should stop with custom exception wrapped (1)' => [new HandlerFailedException(new Envelope(new \stdClass()), [new StopWorkerException()]), true];
+        yield 'it should stop with custom exception wrapped (2)' => [new HandlerFailedException(new Envelope(new \stdClass()), [new \Exception(), new StopWorkerException()]), true];
+        yield 'it should stop with core exception wrapped (1)' => [new HandlerFailedException(new Envelope(new \stdClass()), [$t]), true];
+        yield 'it should stop with core exception wrapped (2)' => [new HandlerFailedException(new Envelope(new \stdClass()), [new \Exception(), $t]), true];
+    }
+
+    /** @dataProvider provideTests */
+    public function test(\Throwable $throwable, bool $shouldStop)
+    {
+        $listener = new StopWorkerOnCustomStopExceptionListener();
+
+        $envelope = new Envelope(new \stdClass());
+        $failedEvent = new WorkerMessageFailedEvent($envelope, 'my_receiver', $throwable);
+
+        $listener->onMessageFailed($failedEvent);
+
+        $worker = $this->createMock(Worker::class);
+        $worker->expects($shouldStop ? $this->once() : $this->never())->method('stop');
+        $runningEvent = new WorkerRunningEvent($worker, false);
+
+        $listener->onWorkerRunning($runningEvent);
+    }
+}