Skip to content

Commit 257ec87

Browse files
lyrixxfabpot
lyrixx
authored and
fabpot
committed
[Messenger] Added StopWorkerException
1 parent 6402572 commit 257ec87

File tree

7 files changed

+167
-2
lines changed

7 files changed

+167
-2
lines changed

‎src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
use Symfony\Component\Messenger\EventListener\DispatchPcntlSignalListener;
2121
use Symfony\Component\Messenger\EventListener\SendFailedMessageForRetryListener;
2222
use Symfony\Component\Messenger\EventListener\SendFailedMessageToFailureTransportListener;
23+
use Symfony\Component\Messenger\EventListener\StopWorkerOnCustomStopExceptionListener;
2324
use Symfony\Component\Messenger\EventListener\StopWorkerOnRestartSignalListener;
2425
use Symfony\Component\Messenger\EventListener\StopWorkerOnSigtermSignalListener;
2526
use Symfony\Component\Messenger\Middleware\AddBusNameStampMiddleware;
@@ -193,6 +194,9 @@
193194
->set('messenger.listener.stop_worker_on_sigterm_signal_listener', StopWorkerOnSigtermSignalListener::class)
194195
->tag('kernel.event_subscriber')
195196

197+
->set('messenger.listener.stop_worker_on_stop_exception_listener', StopWorkerOnCustomStopExceptionListener::class)
198+
->tag('kernel.event_subscriber')
199+
196200
->set('messenger.routable_message_bus', RoutableMessageBus::class)
197201
->args([
198202
abstract_arg('message bus locator'),

‎src/Symfony/Bundle/FrameworkBundle/composer.json

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@
4848
"symfony/http-client": "^4.4|^5.0|^6.0",
4949
"symfony/lock": "^4.4|^5.0|^6.0",
5050
"symfony/mailer": "^5.2|^6.0",
51-
"symfony/messenger": "^5.2|^6.0",
51+
"symfony/messenger": "^5.4|^6.0",
5252
"symfony/mime": "^4.4|^5.0|^6.0",
5353
"symfony/notifier": "^5.3|^6.0",
5454
"symfony/allmysms-notifier": "^5.3|^6.0",
@@ -115,7 +115,7 @@
115115
"symfony/form": "<5.2",
116116
"symfony/lock": "<4.4",
117117
"symfony/mailer": "<5.2",
118-
"symfony/messenger": "<4.4",
118+
"symfony/messenger": "<5.4",
119119
"symfony/mime": "<4.4",
120120
"symfony/property-info": "<4.4",
121121
"symfony/property-access": "<5.3",

‎src/Symfony/Component/Messenger/CHANGELOG.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,11 @@
11
CHANGELOG
22
=========
33

4+
5.4
5+
---
6+
7+
* Add `StopWorkerExceptionInterface` and its implementation `StopWorkerException` to stop the worker.
8+
49
5.3
510
---
611

‎src/Symfony/Component/Messenger/EventListener/StopWorkerOnCustomStopExceptionListener.php

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\EventListener;
13+
14+
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
15+
use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
16+
use Symfony\Component\Messenger\Event\WorkerRunningEvent;
17+
use Symfony\Component\Messenger\Exception\HandlerFailedException;
18+
use Symfony\Component\Messenger\Exception\StopWorkerExceptionInterface;
19+
20+
/**
21+
* @author Grégoire Pineau <lyrixx@lyrixx.info>
22+
*/
23+
class StopWorkerOnCustomStopExceptionListener implements EventSubscriberInterface
24+
{
25+
private $stop = false;
26+
27+
public function onMessageFailed(WorkerMessageFailedEvent $event): void
28+
{
29+
$th = $event->getThrowable();
30+
if ($th instanceof StopWorkerExceptionInterface) {
31+
$this->stop = true;
32+
}
33+
if ($th instanceof HandlerFailedException) {
34+
foreach ($th->getNestedExceptions() as $e) {
35+
if ($e instanceof StopWorkerExceptionInterface) {
36+
$this->stop = true;
37+
break;
38+
}
39+
}
40+
}
41+
}
42+
43+
public function onWorkerRunning(WorkerRunningEvent $event): void
44+
{
45+
if ($this->stop) {
46+
$event->getWorker()->stop();
47+
}
48+
}
49+
50+
public static function getSubscribedEvents(): array
51+
{
52+
return [
53+
WorkerMessageFailedEvent::class => 'onMessageFailed',
54+
WorkerRunningEvent::class => 'onWorkerRunning',
55+
];
56+
}
57+
}

‎src/Symfony/Component/Messenger/Exception/StopWorkerException.php

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\Exception;
13+
14+
/**
15+
* @author Grégoire Pineau <lyrixx@lyrixx.info>
16+
*/
17+
class StopWorkerException extends RuntimeException implements StopWorkerExceptionInterface
18+
{
19+
public function __construct(string $message = 'Worker should stop.', ?\Throwable $previous = null)
20+
{
21+
parent::__construct($message, 0, $previous);
22+
}
23+
}

‎src/Symfony/Component/Messenger/Exception/StopWorkerExceptionInterface.php

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\Exception;
13+
14+
/**
15+
* @author Grégoire Pineau <lyrixx@lyrixx.info>
16+
*/
17+
interface StopWorkerExceptionInterface extends \Throwable
18+
{
19+
}

‎src/Symfony/Component/Messenger/Tests/EventListener/StopWorkerOnCustomStopExceptionListenerTest.php

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\Messenger\Tests\EventListener;
13+
14+
use PHPUnit\Framework\TestCase;
15+
use Symfony\Component\Messenger\Envelope;
16+
use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
17+
use Symfony\Component\Messenger\Event\WorkerRunningEvent;
18+
use Symfony\Component\Messenger\EventListener\StopWorkerOnCustomStopExceptionListener;
19+
use Symfony\Component\Messenger\Exception\HandlerFailedException;
20+
use Symfony\Component\Messenger\Exception\StopWorkerException;
21+
use Symfony\Component\Messenger\Exception\StopWorkerExceptionInterface;
22+
use Symfony\Component\Messenger\Worker;
23+
24+
class StopWorkerOnCustomStopExceptionListenerTest extends TestCase
25+
{
26+
public function provideTests(): \Generator
27+
{
28+
yield 'it should not stop (1)' => [new \Exception(), false];
29+
yield 'it should not stop (2)' => [new HandlerFailedException(new Envelope(new \stdClass()), [new \Exception()]), false];
30+
31+
$t = new class() extends \Exception implements StopWorkerExceptionInterface {};
32+
yield 'it should stop with custom exception' => [$t, true];
33+
yield 'it should stop with core exception' => [new StopWorkerException(), true];
34+
35+
yield 'it should stop with custom exception wrapped (1)' => [new HandlerFailedException(new Envelope(new \stdClass()), [new StopWorkerException()]), true];
36+
yield 'it should stop with custom exception wrapped (2)' => [new HandlerFailedException(new Envelope(new \stdClass()), [new \Exception(), new StopWorkerException()]), true];
37+
yield 'it should stop with core exception wrapped (1)' => [new HandlerFailedException(new Envelope(new \stdClass()), [$t]), true];
38+
yield 'it should stop with core exception wrapped (2)' => [new HandlerFailedException(new Envelope(new \stdClass()), [new \Exception(), $t]), true];
39+
}
40+
41+
/** @dataProvider provideTests */
42+
public function test(\Throwable $throwable, bool $shouldStop)
43+
{
44+
$listener = new StopWorkerOnCustomStopExceptionListener();
45+
46+
$envelope = new Envelope(new \stdClass());
47+
$failedEvent = new WorkerMessageFailedEvent($envelope, 'my_receiver', $throwable);
48+
49+
$listener->onMessageFailed($failedEvent);
50+
51+
$worker = $this->createMock(Worker::class);
52+
$worker->expects($shouldStop ? $this->once() : $this->never())->method('stop');
53+
$runningEvent = new WorkerRunningEvent($worker, false);
54+
55+
$listener->onWorkerRunning($runningEvent);
56+
}
57+
}

0 commit comments

Comments
 (0)