From 53ca49e33c0e64c72308fa7ade1eb211cd4920e9 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Wed, 28 Oct 2026 02:07:16 +0300 Subject: [PATCH 1/3] Fix #347: Install SignalLoop signal handlers only while consuming and restore previous ones --- CHANGELOG.md | 1 + docs/guide/en/loops.md | 11 ++ src/Cli/ScopedLoopInterface.php | 25 ++++ src/Cli/SignalLoop.php | 83 +++++++++++-- src/Command/ListenAllCommand.php | 27 +++-- src/QueueConsumer.php | 41 +++++-- tests/Unit/Cli/SignalLoopTest.php | 127 +++++++++++++++++++- tests/Unit/Command/ListenAllCommandTest.php | 28 +++++ tests/Unit/QueueTest.php | 66 ++++++++++ 9 files changed, 380 insertions(+), 29 deletions(-) create mode 100644 src/Cli/ScopedLoopInterface.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c9589ee..58b0792e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,3 +3,4 @@ ## 1.0.0 under development - Initial release. +- Bug #347: `SignalLoop` installs signal handlers only while messages are consumed and restores the previous ones afterwards, instead of replacing them when it is created (@samdark) diff --git a/docs/guide/en/loops.md b/docs/guide/en/loops.md index e20bfb17..5a93f4d5 100644 --- a/docs/guide/en/loops.md +++ b/docs/guide/en/loops.md @@ -26,6 +26,13 @@ It supports: - Pause/resume via `SIGTSTP` and `SIGCONT`. - Optional soft memory limit (see below). +Signal handlers are installed only while messages are consumed: `QueueConsumer::run()`, `QueueConsumer::listen()` +and `queue:listen-all` install them when they start and restore the previous handlers when they finish. Creating +`SignalLoop`, for example when it is resolved from the container, doesn't affect handlers registered by other code, +such as Symfony Console. While a consumer runs, its handlers replace the previous ones, so that the message being +processed is finished before stopping. If `canContinue()` is called outside of a consumer, the handlers are installed +at that moment and kept. + ### `SimpleLoop` `\Yiisoft\Queue\Cli\SimpleLoop` is used by default when `ext-pcntl` is **not** available. @@ -101,3 +108,7 @@ Implement `LoopInterface` and encapsulate your own stopping conditions: The only requirement is that `canContinue()` returns `false` when the worker should stop. +If the loop needs to change process-wide state only while messages are consumed, such as installing signal handlers, +implement `ScopedLoopInterface` instead. Its `start()` is called before consuming begins and `finish()` after it ends. +Calls may be nested when several consumers share one loop, so restore the state on the outermost `finish()`. + diff --git a/src/Cli/ScopedLoopInterface.php b/src/Cli/ScopedLoopInterface.php new file mode 100644 index 00000000..84c41e9f --- /dev/null +++ b/src/Cli/ScopedLoopInterface.php @@ -0,0 +1,25 @@ +|null Handlers that were active before ours were installed, by signal. `null` if our + * handlers are not installed. + */ + private ?array $previousHandlers = null; + private int $depth = 0; + private bool $installedByStart = false; + /** * @param int $memorySoftLimit Soft RAM limit in bytes. The loop won't let you continue to execute the program if * soft limit is reached. Zero means no limit. */ - public function __construct(protected int $memorySoftLimit = 0) + public function __construct(protected int $memorySoftLimit = 0) {} + + public function start(): void { - foreach (self::SIGNALS_EXIT as $signal) { - pcntl_signal($signal, fn() => $this->exit = true); + if ($this->depth++ === 0 && $this->previousHandlers === null) { + $this->installHandlers(); + $this->installedByStart = true; } - foreach (self::SIGNALS_SUSPEND as $signal) { - pcntl_signal($signal, fn() => $this->pause = true); + } + + public function finish(): void + { + if ($this->depth === 0) { + return; } - foreach (self::SIGNALS_RESUME as $signal) { - pcntl_signal($signal, fn() => $this->pause = false); + + if (--$this->depth === 0 && $this->installedByStart) { + $this->restoreHandlers(); } } @@ -53,6 +76,10 @@ public function canContinue(): bool protected function dispatchSignals(): bool { + if ($this->previousHandlers === null) { + $this->installHandlers(); + } + pcntl_signal_dispatch(); // Wait for resume signal until the loop is suspended @@ -68,4 +95,44 @@ protected function getMemoryLimit(): int { return $this->memorySoftLimit; } + + private function installHandlers(): void + { + $handlers = []; + foreach (self::SIGNALS_EXIT as $signal) { + $handlers[$signal] = function (): void { + $this->exit = true; + }; + } + foreach (self::SIGNALS_SUSPEND as $signal) { + $handlers[$signal] = function (): void { + $this->pause = true; + }; + } + foreach (self::SIGNALS_RESUME as $signal) { + $handlers[$signal] = function (): void { + $this->pause = false; + }; + } + + $this->previousHandlers = []; + foreach ($handlers as $signal => $handler) { + $this->previousHandlers[$signal] = pcntl_signal_get_handler($signal); + pcntl_signal($signal, $handler); + } + } + + private function restoreHandlers(): void + { + // Deliver signals received so far to our handlers, so that they are not lost. + pcntl_signal_dispatch(); + + foreach ($this->previousHandlers ?? [] as $signal => $handler) { + /** @psalm-var callable():void|int $handler */ + pcntl_signal($signal, $handler); + } + + $this->previousHandlers = null; + $this->installedByStart = false; + } } diff --git a/src/Command/ListenAllCommand.php b/src/Command/ListenAllCommand.php index 6735ae09..f3bdde4d 100644 --- a/src/Command/ListenAllCommand.php +++ b/src/Command/ListenAllCommand.php @@ -11,6 +11,7 @@ use Symfony\Component\Console\Input\InputOption; use Symfony\Component\Console\Output\OutputInterface; use Yiisoft\Queue\Cli\LoopInterface; +use Yiisoft\Queue\Cli\ScopedLoopInterface; use Yiisoft\Queue\Provider\QueueConsumerProviderInterface; #[AsCommand( @@ -86,15 +87,25 @@ protected function execute(InputInterface $input, OutputInterface $output): int $limit = (int) $input->getOption('limit'); - while ($this->loop->canContinue()) { - $hasMessages = false; - foreach ($consumers as $consumer) { - $hasMessages = $consumer->run($limit) > 0 || $hasMessages; - } + if ($this->loop instanceof ScopedLoopInterface) { + $this->loop->start(); + } - if (!$hasMessages && $this->loop->canContinue()) { - /** @psalm-var 0|positive-int $pauseSeconds */ - sleep($pauseSeconds); + try { + while ($this->loop->canContinue()) { + $hasMessages = false; + foreach ($consumers as $consumer) { + $hasMessages = $consumer->run($limit) > 0 || $hasMessages; + } + + if (!$hasMessages && $this->loop->canContinue()) { + /** @psalm-var 0|positive-int $pauseSeconds */ + sleep($pauseSeconds); + } + } + } finally { + if ($this->loop instanceof ScopedLoopInterface) { + $this->loop->finish(); } } diff --git a/src/QueueConsumer.php b/src/QueueConsumer.php index c16d1d97..36bc944d 100644 --- a/src/QueueConsumer.php +++ b/src/QueueConsumer.php @@ -8,6 +8,7 @@ use Psr\Log\LoggerInterface; use Yiisoft\Queue\Adapter\AdapterInterface; use Yiisoft\Queue\Cli\LoopInterface; +use Yiisoft\Queue\Cli\ScopedLoopInterface; use Yiisoft\Queue\Message\MessageInterface; use Yiisoft\Queue\Worker\WorkerInterface; @@ -34,13 +35,18 @@ public function run(int $max = 0): int } $this->logger->debug('Start processing queue messages.'); $count = 0; - $this->adapter->runExisting(function (MessageInterface $message) use (&$count, $max): bool { - if (($max > 0 && $count >= $max) || !$this->handle($message)) { - return false; - } - $count++; - return true; - }); + $this->startLoop(); + try { + $this->adapter->runExisting(function (MessageInterface $message) use (&$count, $max): bool { + if (($max > 0 && $count >= $max) || !$this->handle($message)) { + return false; + } + $count++; + return true; + }); + } finally { + $this->finishLoop(); + } $this->logger->info('Processed {count} queue messages.', ['count' => $count]); return $count; } @@ -52,10 +58,29 @@ public function listen(): void return; } $this->logger->info('Start listening to the queue.'); - $this->adapter->subscribe(fn(MessageInterface $message): bool => $this->handle($message)); + $this->startLoop(); + try { + $this->adapter->subscribe(fn(MessageInterface $message): bool => $this->handle($message)); + } finally { + $this->finishLoop(); + } $this->logger->info('Finish listening to the queue.'); } + private function startLoop(): void + { + if ($this->loop instanceof ScopedLoopInterface) { + $this->loop->start(); + } + } + + private function finishLoop(): void + { + if ($this->loop instanceof ScopedLoopInterface) { + $this->loop->finish(); + } + } + private function handle(MessageInterface $message): bool { if (!$this->loop->canContinue()) { diff --git a/tests/Unit/Cli/SignalLoopTest.php b/tests/Unit/Cli/SignalLoopTest.php index ebf06971..0aafaa61 100644 --- a/tests/Unit/Cli/SignalLoopTest.php +++ b/tests/Unit/Cli/SignalLoopTest.php @@ -9,12 +9,15 @@ use PHPUnit\Framework\Attributes\RequiresPhpExtension; use Yiisoft\Queue\Cli\SignalLoop; +use function array_map; + use const SIGALRM; use const SIGCONT; use const SIGHUP; use const SIGINT; use const SIGTERM; use const SIGTSTP; +use const SIG_DFL; /** * @psalm-suppress UndefinedConstant @@ -23,12 +26,117 @@ #[RequiresPhpExtension('pcntl')] final class SignalLoopTest extends TestCase { + private const SIGNALS = [SIGHUP, SIGINT, SIGTERM, SIGTSTP, SIGCONT]; + + /** + * @var SignalLoop[] + */ + private array $startedLoops = []; + public function setUp(): void { parent::setUp(); pcntl_async_signals(true); } + protected function tearDown(): void + { + foreach ($this->startedLoops as $loop) { + $loop->finish(); + } + $this->startedLoops = []; + foreach (self::SIGNALS as $signal) { + pcntl_signal($signal, SIG_DFL); + } + + parent::tearDown(); + } + + public function testConstructionKeepsExistingHandlers(): void + { + $handler = static function (): void {}; + foreach (self::SIGNALS as $signal) { + pcntl_signal($signal, $handler); + } + + new SignalLoop(); + + foreach (self::SIGNALS as $signal) { + self::assertSame($handler, pcntl_signal_get_handler($signal), "Handler of signal {$signal} was replaced."); + } + } + + public function testFinishRestoresHandlersReplacedByStart(): void + { + $handler = static function (): void {}; + foreach (self::SIGNALS as $signal) { + pcntl_signal($signal, $handler); + } + $loop = new SignalLoop(); + + $loop->start(); + $replaced = array_map(pcntl_signal_get_handler(...), self::SIGNALS); + $loop->finish(); + + foreach ($replaced as $signalHandler) { + self::assertNotSame($handler, $signalHandler); + } + foreach (self::SIGNALS as $signal) { + self::assertSame($handler, pcntl_signal_get_handler($signal), "Handler of signal {$signal} was not restored."); + } + } + + public function testNestedStartRestoresHandlersOnOutermostFinish(): void + { + $handler = static function (): void {}; + pcntl_signal(SIGTERM, $handler); + $loop = new SignalLoop(); + + $loop->start(); + $loop->start(); + $loop->finish(); + $afterInnerFinish = pcntl_signal_get_handler(SIGTERM); + $loop->finish(); + + self::assertNotSame($handler, $afterInnerFinish); + self::assertSame($handler, pcntl_signal_get_handler(SIGTERM)); + } + + public function testExtraFinishDoesNothing(): void + { + $handler = static function (): void {}; + pcntl_signal(SIGTERM, $handler); + $loop = new SignalLoop(); + + $loop->finish(); + + self::assertSame($handler, pcntl_signal_get_handler(SIGTERM)); + } + + public function testExitSignalReceivedWhileStartedIsKeptAfterFinish(): void + { + $loop = new SignalLoop(); + $loop->start(); + posix_kill(getmypid(), SIGTERM); + $loop->finish(); + + $loop->start(); + $result = $loop->canContinue(); + $loop->finish(); + + self::assertFalse($result); + } + + public function testCanContinueWithoutStartInstallsHandlers(): void + { + $loop = new SignalLoop(); + + self::assertTrue($loop->canContinue()); + posix_kill(getmypid(), SIGTERM); + + self::assertFalse($loop->canContinue()); + } + public function testMemoryLimitReached(): void { $loop = new SignalLoop(1); @@ -37,7 +145,7 @@ public function testMemoryLimitReached(): void public function testSuspendAndResume(): void { - $loop = new SignalLoop(0); + $loop = $this->createStartedLoop(); pcntl_signal(SIGALRM, static function (): void { posix_kill(getmypid(), SIGCONT); }); @@ -56,7 +164,7 @@ public function testSuspendAndResume(): void #[DataProvider('exitSignalProvider')] public function testExitSignals(int $signal): void { - $loop = new SignalLoop(0); + $loop = $this->createStartedLoop(); self::assertTrue($loop->canContinue(), 'Loop should continue'); posix_kill(getmypid(), $signal); @@ -73,7 +181,7 @@ public static function exitSignalProvider(): iterable public function testResumeSignal(): void { - $loop = new SignalLoop(0); + $loop = $this->createStartedLoop(); // First suspend the loop posix_kill(getmypid(), SIGTSTP); @@ -91,7 +199,7 @@ public function testResumeSignal(): void public function testMultipleExitSignals(): void { - $loop = new SignalLoop(0); + $loop = $this->createStartedLoop(); // Send multiple exit signals posix_kill(getmypid(), SIGINT); @@ -104,7 +212,7 @@ public function testMultipleExitSignals(): void public function testSuspendOverridesResume(): void { - $loop = new SignalLoop(0); + $loop = $this->createStartedLoop(); // Resume first posix_kill(getmypid(), SIGCONT); @@ -124,4 +232,13 @@ public function testSuspendOverridesResume(): void self::assertTrue($result); self::assertGreaterThan(0.5, $elapsed, 'Loop should wait for resume after suspend'); } + + private function createStartedLoop(): SignalLoop + { + $loop = new SignalLoop(0); + $loop->start(); + $this->startedLoops[] = $loop; + + return $loop; + } } diff --git a/tests/Unit/Command/ListenAllCommandTest.php b/tests/Unit/Command/ListenAllCommandTest.php index 7053c02d..d197e3ec 100644 --- a/tests/Unit/Command/ListenAllCommandTest.php +++ b/tests/Unit/Command/ListenAllCommandTest.php @@ -10,6 +10,7 @@ use Symfony\Component\Console\Output\BufferedOutput; use Symfony\Component\Console\Output\OutputInterface; use Yiisoft\Queue\Cli\LoopInterface; +use Yiisoft\Queue\Cli\ScopedLoopInterface; use Yiisoft\Queue\Command\ListenAllCommand; use Yiisoft\Queue\Provider\PredefinedQueueProvider; use Yiisoft\Queue\QueueConsumerInterface; @@ -82,4 +83,31 @@ public function canContinue(): bool } }; } + + public function testStartsAndFinishesScopedLoopOnce(): void + { + $calls = []; + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->method('run')->willReturnCallback(static function () use (&$calls): int { + $calls[] = 'run'; + return 0; + }); + $loop = $this->createMock(ScopedLoopInterface::class); + $loop->method('start')->willReturnCallback(static function () use (&$calls): void { + $calls[] = 'start'; + }); + $loop->method('finish')->willReturnCallback(static function () use (&$calls): void { + $calls[] = 'finish'; + }); + $loop->method('canContinue')->willReturn(true, false); + $command = new ListenAllCommand(new PredefinedQueueProvider([ + 'consumer' => ['consumer' => $consumer], + ]), $loop); + $input = new ArrayInput([], $command->getNativeDefinition()); + $input->setOption('pause', 0); + + $command->run($input, $this->createMock(OutputInterface::class)); + + self::assertSame(['start', 'run', 'finish'], $calls); + } } diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index a0bc23ca..80aec76c 100644 --- a/tests/Unit/QueueTest.php +++ b/tests/Unit/QueueTest.php @@ -5,6 +5,9 @@ namespace Yiisoft\Queue\Tests\Unit; use Psr\Log\NullLogger; +use RuntimeException; +use Yiisoft\Queue\Adapter\AdapterInterface; +use Yiisoft\Queue\Cli\ScopedLoopInterface; use Yiisoft\Queue\QueueConsumer; use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueProducerInterface; @@ -119,4 +122,67 @@ public function testMessageHandledBeforeStopIsNotHandledAgain(): void self::assertSame(2, $this->executionTimes, 'The message handled before the stop is not handled again.'); self::assertCount(0, $adapter->getMessagesList()); } + + public function testConsumerRunStartsAndFinishesScopedLoop(): void + { + $adapter = new InMemoryAdapter(); + $this->createQueue($adapter)->push(new GenericMessage('simple', null)); + $loop = new RecordingScopedLoop(); + $consumer = new QueueConsumer($this->getWorker(), $loop, new NullLogger(), $adapter); + + $consumer->run(); + + self::assertSame(['start', 'canContinue', 'finish'], $loop->calls); + } + + public function testConsumerListenStartsAndFinishesScopedLoop(): void + { + $adapter = new InMemoryAdapter(); + $this->createQueue($adapter)->push(new GenericMessage('simple', null)); + $loop = new RecordingScopedLoop(); + $consumer = new QueueConsumer($this->getWorker(), $loop, new NullLogger(), $adapter); + + $consumer->listen(); + + self::assertSame(['start', 'canContinue', 'finish'], $loop->calls); + } + + public function testConsumerFinishesScopedLoopWhenAdapterThrows(): void + { + $adapter = $this->createMock(AdapterInterface::class); + $adapter->method('runExisting')->willThrowException(new RuntimeException('Adapter failed.')); + $loop = new RecordingScopedLoop(); + $consumer = new QueueConsumer($this->getWorker(), $loop, new NullLogger(), $adapter); + + try { + $consumer->run(); + } catch (RuntimeException) { + } + + self::assertSame(['start', 'finish'], $loop->calls); + } +} + +final class RecordingScopedLoop implements ScopedLoopInterface +{ + /** + * @var string[] + */ + public array $calls = []; + + public function start(): void + { + $this->calls[] = 'start'; + } + + public function finish(): void + { + $this->calls[] = 'finish'; + } + + public function canContinue(): bool + { + $this->calls[] = 'canContinue'; + return true; + } } From 92b07c7cd24d4d42ae292ea62ad0e7ad2fd21f15 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Fri, 30 Oct 2026 00:56:38 +0300 Subject: [PATCH 2/3] Keep signal handlers installed between queues in `queue:run`, bind `--pause` in test - `RunCommand` accepts an optional loop and starts it once around all queues, so a signal received between two queues sets the exit flag instead of reaching the default handler. - Document and test that handlers installed by `canContinue()` outside of `start()`/`finish()` are kept. - Pass `--pause` through `ArrayInput` parameters, since `Command::run()` re-binds the input. --- docs/guide/en/loops.md | 5 +-- src/Cli/SignalLoop.php | 3 +- src/Command/RunCommand.php | 28 +++++++++++++---- tests/Unit/Cli/SignalLoopTest.php | 15 +++++++++ tests/Unit/Command/ListenAllCommandTest.php | 35 ++++++++++----------- tests/Unit/Command/RunCommandTest.php | 29 +++++++++++++++++ 6 files changed, 88 insertions(+), 27 deletions(-) diff --git a/docs/guide/en/loops.md b/docs/guide/en/loops.md index 5a93f4d5..1938da27 100644 --- a/docs/guide/en/loops.md +++ b/docs/guide/en/loops.md @@ -26,8 +26,9 @@ It supports: - Pause/resume via `SIGTSTP` and `SIGCONT`. - Optional soft memory limit (see below). -Signal handlers are installed only while messages are consumed: `QueueConsumer::run()`, `QueueConsumer::listen()` -and `queue:listen-all` install them when they start and restore the previous handlers when they finish. Creating +Signal handlers are installed only while messages are consumed: `QueueConsumer::run()`, `QueueConsumer::listen()`, +`queue:run` and `queue:listen-all` install them when they start and restore the previous handlers when they finish. +The commands keep them installed between queues. Creating `SignalLoop`, for example when it is resolved from the container, doesn't affect handlers registered by other code, such as Symfony Console. While a consumer runs, its handlers replace the previous ones, so that the message being processed is finished before stopping. If `canContinue()` is called outside of a consumer, the handlers are installed diff --git a/src/Cli/SignalLoop.php b/src/Cli/SignalLoop.php index 411520ab..6d913f0d 100644 --- a/src/Cli/SignalLoop.php +++ b/src/Cli/SignalLoop.php @@ -15,7 +15,8 @@ * * Signal handlers are installed by {@see start()} and the previous ones are restored by {@see finish()}, so creating * the loop doesn't affect handlers registered by other code, such as Symfony Console. If {@see canContinue()} is - * called outside of {@see start()} and {@see finish()}, the handlers are installed then and kept. + * called outside of {@see start()} and {@see finish()}, the handlers are installed then and kept, also through any + * later {@see start()} and {@see finish()}: the code using the loop that way relies on them until it stops. */ final class SignalLoop implements ScopedLoopInterface { diff --git a/src/Command/RunCommand.php b/src/Command/RunCommand.php index 648eefed..b57d676d 100644 --- a/src/Command/RunCommand.php +++ b/src/Command/RunCommand.php @@ -10,6 +10,8 @@ use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Input\InputOption; use Symfony\Component\Console\Output\OutputInterface; +use Yiisoft\Queue\Cli\LoopInterface; +use Yiisoft\Queue\Cli\ScopedLoopInterface; use Yiisoft\Queue\Provider\QueueConsumerProviderInterface; #[AsCommand( @@ -20,6 +22,7 @@ final class RunCommand extends Command { public function __construct( private readonly QueueConsumerProviderInterface $queueProvider, + private readonly ?LoopInterface $loop = null, ) { parent::__construct(); } @@ -50,14 +53,27 @@ protected function execute(InputInterface $input, OutputInterface $output): int $queueNames = $this->queueProvider->getConsumerQueueNames(); } - /** @var string $queueName */ - foreach ($queueNames as $queueName) { - $queueConsumer = $this->queueProvider->getConsumer($queueName); + $limit = (int) $input->getOption('limit'); - $output->write("Processing queue $queueName... "); - $count = $queueConsumer->run((int) $input->getOption('limit')); + // One scope for all queues, so that a signal received between two of them doesn't reach the default handler. + if ($this->loop instanceof ScopedLoopInterface) { + $this->loop->start(); + } + + try { + /** @var string $queueName */ + foreach ($queueNames as $queueName) { + $queueConsumer = $this->queueProvider->getConsumer($queueName); + + $output->write("Processing queue $queueName... "); + $count = $queueConsumer->run($limit); - $output->writeln("Messages processed: $count."); + $output->writeln("Messages processed: $count."); + } + } finally { + if ($this->loop instanceof ScopedLoopInterface) { + $this->loop->finish(); + } } return Command::SUCCESS; diff --git a/tests/Unit/Cli/SignalLoopTest.php b/tests/Unit/Cli/SignalLoopTest.php index 0aafaa61..052e16d1 100644 --- a/tests/Unit/Cli/SignalLoopTest.php +++ b/tests/Unit/Cli/SignalLoopTest.php @@ -137,6 +137,21 @@ public function testCanContinueWithoutStartInstallsHandlers(): void self::assertFalse($loop->canContinue()); } + public function testFinishKeepsHandlersInstalledByCanContinueBeforeStart(): void + { + $handler = static function (): void {}; + pcntl_signal(SIGTERM, $handler); + $loop = new SignalLoop(); + $loop->canContinue(); + $installed = pcntl_signal_get_handler(SIGTERM); + + $loop->start(); + $loop->finish(); + + self::assertNotSame($handler, $installed); + self::assertSame($installed, pcntl_signal_get_handler(SIGTERM)); + } + public function testMemoryLimitReached(): void { $loop = new SignalLoop(1); diff --git a/tests/Unit/Command/ListenAllCommandTest.php b/tests/Unit/Command/ListenAllCommandTest.php index d197e3ec..4d14f737 100644 --- a/tests/Unit/Command/ListenAllCommandTest.php +++ b/tests/Unit/Command/ListenAllCommandTest.php @@ -69,21 +69,6 @@ public function testPausesWhenThereAreNoMessages(): void self::assertGreaterThanOrEqual(1.0, (hrtime(true) - $start) / 1_000_000_000, 'An empty queue is polled once per pause.'); } - /** - * @return LoopInterface A loop which allows the given number of checks and then asks to stop. - */ - private function createLoop(int $continueTimes): LoopInterface - { - return new class ($continueTimes) implements LoopInterface { - public function __construct(private int $continueTimes) {} - - public function canContinue(): bool - { - return $this->continueTimes-- > 0; - } - }; - } - public function testStartsAndFinishesScopedLoopOnce(): void { $calls = []; @@ -99,15 +84,29 @@ public function testStartsAndFinishesScopedLoopOnce(): void $loop->method('finish')->willReturnCallback(static function () use (&$calls): void { $calls[] = 'finish'; }); - $loop->method('canContinue')->willReturn(true, false); + $loop->method('canContinue')->willReturn(true, false, false); $command = new ListenAllCommand(new PredefinedQueueProvider([ 'consumer' => ['consumer' => $consumer], ]), $loop); - $input = new ArrayInput([], $command->getNativeDefinition()); - $input->setOption('pause', 0); + $input = new ArrayInput(['--pause' => 0], $command->getNativeDefinition()); $command->run($input, $this->createMock(OutputInterface::class)); self::assertSame(['start', 'run', 'finish'], $calls); } + + /** + * @return LoopInterface A loop which allows the given number of checks and then asks to stop. + */ + private function createLoop(int $continueTimes): LoopInterface + { + return new class ($continueTimes) implements LoopInterface { + public function __construct(private int $continueTimes) {} + + public function canContinue(): bool + { + return $this->continueTimes-- > 0; + } + }; + } } diff --git a/tests/Unit/Command/RunCommandTest.php b/tests/Unit/Command/RunCommandTest.php index 133cdd83..bc2f6af4 100644 --- a/tests/Unit/Command/RunCommandTest.php +++ b/tests/Unit/Command/RunCommandTest.php @@ -7,6 +7,7 @@ use PHPUnit\Framework\TestCase; use Symfony\Component\Console\Input\StringInput; use Symfony\Component\Console\Output\OutputInterface; +use Yiisoft\Queue\Cli\ScopedLoopInterface; use Yiisoft\Queue\Command\RunCommand; use Yiisoft\Queue\DefaultQueue; use Yiisoft\Queue\Provider\PredefinedQueueProvider; @@ -36,4 +37,32 @@ public function testDefaultRunSkipsProducerOnlyQueues(): void ])); self::assertSame(0, $command->run(new StringInput(''), $this->createMock(OutputInterface::class))); } + + public function testStartsAndFinishesScopedLoopOnceForAllQueues(): void + { + $calls = []; + $createConsumer = function (string $name) use (&$calls): QueueConsumerInterface { + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->method('run')->willReturnCallback(static function () use (&$calls, $name): int { + $calls[] = "run $name"; + return 0; + }); + return $consumer; + }; + $loop = $this->createMock(ScopedLoopInterface::class); + $loop->method('start')->willReturnCallback(static function () use (&$calls): void { + $calls[] = 'start'; + }); + $loop->method('finish')->willReturnCallback(static function () use (&$calls): void { + $calls[] = 'finish'; + }); + $command = new RunCommand(new PredefinedQueueProvider([ + 'first' => ['consumer' => $createConsumer('first')], + 'second' => ['consumer' => $createConsumer('second')], + ]), $loop); + + $command->run(new StringInput('first second'), $this->createMock(OutputInterface::class)); + + self::assertSame(['start', 'run first', 'run second', 'finish'], $calls); + } } From 8af7835d60c00747bff84594f28e9a61f948e39c Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Sat, 31 Oct 2026 01:42:34 +0300 Subject: [PATCH 3/3] Clarify signal loop usage and behavior tests --- docs/guide/en/loops.md | 51 ++++- src/Cli/ScopedLoopInterface.php | 6 +- src/Cli/SignalLoop.php | 16 +- tests/Unit/Command/ListenAllCommandTest.php | 42 +--- tests/Unit/Command/RunCommandTest.php | 29 --- tests/Unit/QueueTest.php | 85 +------- tests/Unit/SignalHandlingTest.php | 208 ++++++++++++++++++++ tests/Unit/TestQueue.php | 10 + 8 files changed, 283 insertions(+), 164 deletions(-) create mode 100644 tests/Unit/SignalHandlingTest.php create mode 100644 tests/Unit/TestQueue.php diff --git a/docs/guide/en/loops.md b/docs/guide/en/loops.md index 73800bee..0e925c89 100644 --- a/docs/guide/en/loops.md +++ b/docs/guide/en/loops.md @@ -27,13 +27,18 @@ It supports: - Pause/resume via `SIGTSTP` and `SIGCONT`. - Optional soft memory limit (see below). -Signal handlers are installed only while messages are consumed: `QueueConsumer::run()`, `QueueConsumer::listen()`, -`queue:run` and `queue:listen-all` install them when they start and restore the previous handlers when they finish. -The commands keep them installed between queues. Creating -`SignalLoop`, for example when it is resolved from the container, doesn't affect handlers registered by other code, -such as Symfony Console. While a consumer runs, its handlers replace the previous ones, so that the message being -processed is finished before stopping. If `canContinue()` is called outside of a consumer, the handlers are installed -at that moment and kept. +To stop a worker gracefully, send `SIGTERM` to its process, for example: + +```shell +kill -TERM +``` + +The worker finishes the current message and leaves the remaining messages in the queue for the next worker. +Use `SIGTSTP` to pause processing and `SIGCONT` to resume it. + +You can create or resolve `SignalLoop` in another console command without changing that command's signal handling. +Existing application handlers are restored when queue processing ends, including when processing fails. +While a queue worker is running, its graceful-shutdown behavior takes precedence over application handlers. ### `SimpleLoop` @@ -99,6 +104,31 @@ use Yiisoft\Queue\Cli\SignalLoop; $loop = new SignalLoop(memorySoftLimit: 256 * 1024 * 1024); ``` +When creating `RunCommand` or `ListenAllCommand` manually, pass the same loop instance to the command and its consumers. +This keeps graceful shutdown active between queues. For example, with your configured consumer provider: + +```php +use Yiisoft\Queue\Command\RunCommand; + +$command = new RunCommand($consumerProvider, $loop); +``` + +For your own execution loop, use `start()` and `finish()` to restore application signal handlers when work ends: + +```php +$loop->start(); +try { + while ($loop->canContinue()) { + // Process one unit of work. + } +} finally { + $loop->finish(); +} +``` + +If you call `canContinue()` without `start()`, the loop keeps its signal handlers for the rest of the process. +Use the paired calls above when other application code needs to handle signals after your work ends. + ## Writing a custom loop Implement `LoopInterface` and encapsulate your own stopping conditions: @@ -110,7 +140,8 @@ Implement `LoopInterface` and encapsulate your own stopping conditions: The only requirement is that `canContinue()` returns `false` when the worker should stop. -If the loop needs to change process-wide state only while messages are consumed, such as installing signal handlers, -implement `ScopedLoopInterface` instead. Its `start()` is called before consuming begins and `finish()` after it ends. -Calls may be nested when several consumers share one loop, so restore the state on the outermost `finish()`. +Implement `ScopedLoopInterface` when your custom loop needs setup and cleanup around message processing. +Put setup in `start()` and cleanup in `finish()`; consumers and commands call these methods for you, +even if processing throws. Support nested calls when consumers share a loop with a command: keep the loop active +until the matching outermost `finish()`. diff --git a/src/Cli/ScopedLoopInterface.php b/src/Cli/ScopedLoopInterface.php index 84c41e9f..53985eb4 100644 --- a/src/Cli/ScopedLoopInterface.php +++ b/src/Cli/ScopedLoopInterface.php @@ -7,9 +7,9 @@ /** * A loop that changes process-wide state, such as signal handlers, only while messages are consumed. * - * {@see start()} is called before consuming starts and {@see finish()} after it ends. Calls may be nested, for example + * {@see ScopedLoopInterface::start()} is called before consuming starts and {@see ScopedLoopInterface::finish()} after it ends. Calls may be nested, for example * when `queue:listen-all` runs consumers that share the same loop: the state is changed by the outermost - * {@see start()} and restored by the matching {@see finish()}. + * {@see ScopedLoopInterface::start()} and restored by the matching {@see ScopedLoopInterface::finish()}. */ interface ScopedLoopInterface extends LoopInterface { @@ -19,7 +19,7 @@ interface ScopedLoopInterface extends LoopInterface public function start(): void; /** - * Restores what {@see start()} changed after consuming messages. + * Restores what {@see ScopedLoopInterface::start()} changed after consuming messages. */ public function finish(): void; } diff --git a/src/Cli/SignalLoop.php b/src/Cli/SignalLoop.php index 6d913f0d..e3bc9911 100644 --- a/src/Cli/SignalLoop.php +++ b/src/Cli/SignalLoop.php @@ -13,10 +13,10 @@ /** * A loop that stops on `SIGHUP`, `SIGINT` and `SIGTERM`, and pauses on `SIGTSTP` until `SIGCONT`. * - * Signal handlers are installed by {@see start()} and the previous ones are restored by {@see finish()}, so creating - * the loop doesn't affect handlers registered by other code, such as Symfony Console. If {@see canContinue()} is - * called outside of {@see start()} and {@see finish()}, the handlers are installed then and kept, also through any - * later {@see start()} and {@see finish()}: the code using the loop that way relies on them until it stops. + * Pass the same loop to your consumers and commands to finish the current message before stopping. + * Creating a loop preserves existing application signal handlers. Consumers restore them after processing ends. + * For a custom execution loop, pair {@see SignalLoop::start()} with {@see SignalLoop::finish()} in a `finally` block. + * Calling {@see SignalLoop::canContinue()} without starting a scope keeps the handlers for the rest of the process. */ final class SignalLoop implements ScopedLoopInterface { @@ -44,7 +44,9 @@ public function __construct(protected int $memorySoftLimit = 0) {} public function start(): void { - if ($this->depth++ === 0 && $this->previousHandlers === null) { + $this->depth++; + + if ($this->depth === 1 && $this->previousHandlers === null) { $this->installHandlers(); $this->installedByStart = true; } @@ -56,7 +58,9 @@ public function finish(): void return; } - if (--$this->depth === 0 && $this->installedByStart) { + $this->depth--; + + if ($this->depth === 0 && $this->installedByStart) { $this->restoreHandlers(); } } diff --git a/tests/Unit/Command/ListenAllCommandTest.php b/tests/Unit/Command/ListenAllCommandTest.php index 4d14f737..a2d6b214 100644 --- a/tests/Unit/Command/ListenAllCommandTest.php +++ b/tests/Unit/Command/ListenAllCommandTest.php @@ -10,7 +10,6 @@ use Symfony\Component\Console\Output\BufferedOutput; use Symfony\Component\Console\Output\OutputInterface; use Yiisoft\Queue\Cli\LoopInterface; -use Yiisoft\Queue\Cli\ScopedLoopInterface; use Yiisoft\Queue\Command\ListenAllCommand; use Yiisoft\Queue\Provider\PredefinedQueueProvider; use Yiisoft\Queue\QueueConsumerInterface; @@ -69,44 +68,19 @@ public function testPausesWhenThereAreNoMessages(): void self::assertGreaterThanOrEqual(1.0, (hrtime(true) - $start) / 1_000_000_000, 'An empty queue is polled once per pause.'); } - public function testStartsAndFinishesScopedLoopOnce(): void - { - $calls = []; - $consumer = $this->createMock(QueueConsumerInterface::class); - $consumer->method('run')->willReturnCallback(static function () use (&$calls): int { - $calls[] = 'run'; - return 0; - }); - $loop = $this->createMock(ScopedLoopInterface::class); - $loop->method('start')->willReturnCallback(static function () use (&$calls): void { - $calls[] = 'start'; - }); - $loop->method('finish')->willReturnCallback(static function () use (&$calls): void { - $calls[] = 'finish'; - }); - $loop->method('canContinue')->willReturn(true, false, false); - $command = new ListenAllCommand(new PredefinedQueueProvider([ - 'consumer' => ['consumer' => $consumer], - ]), $loop); - $input = new ArrayInput(['--pause' => 0], $command->getNativeDefinition()); - - $command->run($input, $this->createMock(OutputInterface::class)); - - self::assertSame(['start', 'run', 'finish'], $calls); - } - /** * @return LoopInterface A loop which allows the given number of checks and then asks to stop. */ private function createLoop(int $continueTimes): LoopInterface { - return new class ($continueTimes) implements LoopInterface { - public function __construct(private int $continueTimes) {} + $loop = $this->createMock(LoopInterface::class); + $loop->method('canContinue')->willReturnCallback(static function () use (&$continueTimes): bool { + $canContinue = $continueTimes > 0; + $continueTimes--; + + return $canContinue; + }); - public function canContinue(): bool - { - return $this->continueTimes-- > 0; - } - }; + return $loop; } } diff --git a/tests/Unit/Command/RunCommandTest.php b/tests/Unit/Command/RunCommandTest.php index bc2f6af4..133cdd83 100644 --- a/tests/Unit/Command/RunCommandTest.php +++ b/tests/Unit/Command/RunCommandTest.php @@ -7,7 +7,6 @@ use PHPUnit\Framework\TestCase; use Symfony\Component\Console\Input\StringInput; use Symfony\Component\Console\Output\OutputInterface; -use Yiisoft\Queue\Cli\ScopedLoopInterface; use Yiisoft\Queue\Command\RunCommand; use Yiisoft\Queue\DefaultQueue; use Yiisoft\Queue\Provider\PredefinedQueueProvider; @@ -37,32 +36,4 @@ public function testDefaultRunSkipsProducerOnlyQueues(): void ])); self::assertSame(0, $command->run(new StringInput(''), $this->createMock(OutputInterface::class))); } - - public function testStartsAndFinishesScopedLoopOnceForAllQueues(): void - { - $calls = []; - $createConsumer = function (string $name) use (&$calls): QueueConsumerInterface { - $consumer = $this->createMock(QueueConsumerInterface::class); - $consumer->method('run')->willReturnCallback(static function () use (&$calls, $name): int { - $calls[] = "run $name"; - return 0; - }); - return $consumer; - }; - $loop = $this->createMock(ScopedLoopInterface::class); - $loop->method('start')->willReturnCallback(static function () use (&$calls): void { - $calls[] = 'start'; - }); - $loop->method('finish')->willReturnCallback(static function () use (&$calls): void { - $calls[] = 'finish'; - }); - $command = new RunCommand(new PredefinedQueueProvider([ - 'first' => ['consumer' => $createConsumer('first')], - 'second' => ['consumer' => $createConsumer('second')], - ]), $loop); - - $command->run(new StringInput('first second'), $this->createMock(OutputInterface::class)); - - self::assertSame(['start', 'run first', 'run second', 'finish'], $calls); - } } diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index 9ceb933a..6f74722b 100644 --- a/tests/Unit/QueueTest.php +++ b/tests/Unit/QueueTest.php @@ -4,11 +4,7 @@ namespace Yiisoft\Queue\Tests\Unit; -use Closure; use Psr\Log\NullLogger; -use RuntimeException; -use Yiisoft\Queue\Adapter\AdapterInterface; -use Yiisoft\Queue\Cli\ScopedLoopInterface; use Yiisoft\Queue\QueueConsumer; use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueProducerInterface; @@ -22,11 +18,6 @@ use function count; -enum TestQueue: string -{ - case HIGH_PRIORITY = 'high-priority'; -} - final class QueueTest extends TestCase { public function testProducerContract(): void @@ -124,45 +115,6 @@ public function testMessageHandledBeforeStopIsNotHandledAgain(): void self::assertCount(0, $adapter->getMessagesList()); } - public function testConsumerRunStartsAndFinishesScopedLoop(): void - { - $adapter = new InMemoryAdapter(); - $this->createQueue($adapter)->push(new GenericMessage('simple', null)); - $loop = new RecordingScopedLoop(); - $consumer = new QueueConsumer($this->getWorker(), $loop, new NullLogger(), $adapter); - - $consumer->run(); - - self::assertSame(['start', 'canContinue', 'finish'], $loop->calls); - } - - public function testConsumerListenStartsAndFinishesScopedLoop(): void - { - $adapter = new InMemoryAdapter(); - $this->createQueue($adapter)->push(new GenericMessage('simple', null)); - $loop = new RecordingScopedLoop(); - $consumer = new QueueConsumer($this->getWorker(), $loop, new NullLogger(), $adapter); - - $consumer->listen(); - - self::assertSame(['start', 'canContinue', 'finish'], $loop->calls); - } - - public function testConsumerFinishesScopedLoopWhenAdapterThrows(): void - { - $adapter = $this->createMock(AdapterInterface::class); - $adapter->method('runExisting')->willThrowException(new RuntimeException('Adapter failed.')); - $loop = new RecordingScopedLoop(); - $consumer = new QueueConsumer($this->getWorker(), $loop, new NullLogger(), $adapter); - - try { - $consumer->run(); - } catch (RuntimeException) { - } - - self::assertSame(['start', 'finish'], $loop->calls); - } - public function testConsumerListenReportsMessageProcessedBeforeLoopStops(): void { $adapter = new InMemoryAdapter(); @@ -193,40 +145,9 @@ public function testConsumerListenStopsBeforeProcessingAnotherMessage(): void private function createStopAfterFirstExecutionLoop(): LoopInterface { - return new class (fn(): int => $this->executionTimes) implements LoopInterface { - /** - * @param Closure(): int $executionTimes - */ - public function __construct(private readonly Closure $executionTimes) {} - - public function canContinue(): bool - { - return ($this->executionTimes)() === 0; - } - }; - } -} - -final class RecordingScopedLoop implements ScopedLoopInterface -{ - /** - * @var string[] - */ - public array $calls = []; - - public function start(): void - { - $this->calls[] = 'start'; - } - - public function finish(): void - { - $this->calls[] = 'finish'; - } + $loop = $this->createMock(LoopInterface::class); + $loop->method('canContinue')->willReturnCallback(fn(): bool => $this->executionTimes === 0); - public function canContinue(): bool - { - $this->calls[] = 'canContinue'; - return true; + return $loop; } } diff --git a/tests/Unit/SignalHandlingTest.php b/tests/Unit/SignalHandlingTest.php new file mode 100644 index 00000000..32572709 --- /dev/null +++ b/tests/Unit/SignalHandlingTest.php @@ -0,0 +1,208 @@ +previousHandlers[$signal] = pcntl_signal_get_handler($signal); + } + $this->previousAsyncSignals = pcntl_async_signals(true); + } + + protected function tearDown(): void + { + foreach ($this->previousHandlers as $signal => $handler) { + pcntl_signal($signal, $handler); + } + pcntl_async_signals($this->previousAsyncSignals); + + parent::tearDown(); + } + + #[DataProvider('consumerMethodProvider')] + public function testShutdownFinishesCurrentMessageAndLeavesNextMessageQueued(string $method): void + { + $applicationSignals = []; + pcntl_signal(SIGTERM, static function (int $signal) use (&$applicationSignals): void { + $applicationSignals[] = $signal; + }); + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $first = IdEnvelope::fromMessage($producer->push(new GenericMessage('stop', null))); + $second = IdEnvelope::fromMessage($producer->push(new GenericMessage('stop', null))); + $finishedMessages = 0; + $worker = $this->createMock(WorkerInterface::class); + $worker->expects($this->once())->method('process')->willReturnCallback( + static function () use (&$finishedMessages): void { + posix_kill(getmypid(), SIGTERM); + $finishedMessages++; + }, + ); + $consumer = new QueueConsumer($worker, new SignalLoop(), new NullLogger(), $adapter); + + $consumer->$method(); + posix_kill(getmypid(), SIGTERM); + + self::assertSame(1, $finishedMessages); + self::assertSame(MessageStatus::DONE, $adapter->status($first->getId())); + self::assertSame(MessageStatus::WAITING, $adapter->status($second->getId())); + self::assertSame([SIGTERM], $applicationSignals, 'Application handles signals again after consuming ends.'); + } + + #[DataProvider('consumerMethodProvider')] + public function testProcessingFailureRestoresApplicationSignalHandling(string $method): void + { + $applicationSignals = []; + pcntl_signal(SIGTERM, static function (int $signal) use (&$applicationSignals): void { + $applicationSignals[] = $signal; + }); + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $message = IdEnvelope::fromMessage($producer->push(new GenericMessage('failing', null))); + $failure = new RuntimeException('Worker failed.'); + $worker = $this->createMock(WorkerInterface::class); + $worker->method('process')->willThrowException($failure); + $consumer = new QueueConsumer($worker, new SignalLoop(), new NullLogger(), $adapter); + + try { + $consumer->$method(); + self::fail('The processing failure must reach the caller.'); + } catch (RuntimeException $exception) { + self::assertSame($failure, $exception); + } + posix_kill(getmypid(), SIGTERM); + + self::assertSame([SIGTERM], $applicationSignals); + self::assertSame(MessageStatus::WAITING, $adapter->status($message->getId())); + } + + public static function consumerMethodProvider(): iterable + { + yield 'run' => ['run']; + yield 'listen' => ['listen']; + } + + /** + * @param class-string|class-string $commandClass + */ + #[DataProvider('commandProvider')] + public function testSignalBetweenQueuesStopsBeforeTheNextMessage(string $commandClass): void + { + $applicationSignals = []; + pcntl_signal(SIGTERM, static function (int $signal) use (&$applicationSignals): void { + $applicationSignals[] = $signal; + }); + $loop = new SignalLoop(); + $firstAdapter = new InMemoryAdapter(); + $firstMessage = IdEnvelope::fromMessage( + $this->createQueue($firstAdapter)->push(new GenericMessage('stop', null)), + ); + $secondAdapter = new InMemoryAdapter(); + $secondMessage = IdEnvelope::fromMessage( + $this->createQueue($secondAdapter)->push(new GenericMessage('later', null)), + ); + $firstWorker = $this->createMock(WorkerInterface::class); + $firstWorker->expects($this->once())->method('process'); + $firstConsumer = new QueueConsumer($firstWorker, $loop, new NullLogger(), $firstAdapter); + $consumerAtQueueBoundary = $this->createMock(QueueConsumerInterface::class); + $consumerAtQueueBoundary->method('run')->willReturnCallback(static function () use ($firstConsumer): int { + $count = $firstConsumer->run(); + posix_kill(getmypid(), SIGTERM); + + return $count; + }); + $secondWorker = $this->createMock(WorkerInterface::class); + $secondWorker->expects($this->never())->method('process'); + $provider = new PredefinedQueueProvider([ + 'first' => ['consumer' => $consumerAtQueueBoundary], + 'second' => ['consumer' => new QueueConsumer($secondWorker, $loop, new NullLogger(), $secondAdapter)], + ]); + $command = new $commandClass($provider, $loop); + $input = new ArrayInput([], $command->getNativeDefinition()); + + $result = $command->run($input, new BufferedOutput()); + posix_kill(getmypid(), SIGTERM); + + self::assertSame(0, $result); + self::assertSame(MessageStatus::DONE, $firstAdapter->status($firstMessage->getId())); + self::assertSame(MessageStatus::WAITING, $secondAdapter->status($secondMessage->getId())); + self::assertSame([SIGTERM], $applicationSignals); + } + + /** + * @param class-string|class-string $commandClass + */ + #[DataProvider('commandProvider')] + public function testCommandFailureRestoresApplicationSignalHandling(string $commandClass): void + { + $applicationSignals = []; + pcntl_signal(SIGTERM, static function (int $signal) use (&$applicationSignals): void { + $applicationSignals[] = $signal; + }); + $adapter = new InMemoryAdapter(); + $this->createQueue($adapter)->push(new GenericMessage('failing', null)); + $failure = new RuntimeException('Worker failed.'); + $worker = $this->createMock(WorkerInterface::class); + $worker->method('process')->willThrowException($failure); + $loop = new SignalLoop(); + $consumer = new QueueConsumer($worker, $loop, new NullLogger(), $adapter); + $command = new $commandClass(new PredefinedQueueProvider(['queue' => ['consumer' => $consumer]]), $loop); + $input = new ArrayInput([], $command->getNativeDefinition()); + + try { + $command->run($input, new BufferedOutput()); + self::fail('The processing failure must reach the caller.'); + } catch (RuntimeException $exception) { + self::assertSame($failure, $exception); + } + posix_kill(getmypid(), SIGTERM); + + self::assertSame([SIGTERM], $applicationSignals); + } + + /** + * @return iterable|class-string}> + */ + public static function commandProvider(): iterable + { + yield 'queue:run' => [RunCommand::class]; + yield 'queue:listen-all' => [ListenAllCommand::class]; + } +} diff --git a/tests/Unit/TestQueue.php b/tests/Unit/TestQueue.php new file mode 100644 index 00000000..24748985 --- /dev/null +++ b/tests/Unit/TestQueue.php @@ -0,0 +1,10 @@ +