From e3c478b04963e2d30402b2259b3f1faa3af8da8a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Viktor=20Ma=C5=A1=C3=AD=C4=8Dek?= <viktor@appsdevteam.com>
Date: Tue, 23 Jun 2026 10:54:32 +0200
Subject: [PATCH 1/2] Implement label-specific DIE queues for targeted consumer
 restart

- Add Manager methods getTopPriorityName() and includeTopPriority() to support per-label DIE queues (e.g., 0_worker)
- Update Consumer and Producer interfaces to accept optional consumerLabel parameter
- Add -l/--label option to consume and reload-consumers commands
- Enhance reload-consumers to send DIE per label with -l option (e.g., -l label1,label2) while maintaining backward compatibility (number of DIE messages per label/queue)
- Update Producer helper signatures to match interface changes
- Add comprehensive unit and integration tests for label queue naming and reload logic
- Update README and CLAUDE.md with new functionality and usage guidance
---
 CLAUDE.md                                     |  6 +-
 README.md                                     |  5 +-
 src/Broker/Consumer.php                       |  2 +-
 src/Broker/PhpAmqpLib/Consumer.php            |  5 +-
 src/Broker/PhpAmqpLib/Manager.php             | 34 ++++++-
 src/Broker/PhpAmqpLib/Producer.php            |  6 +-
 src/Broker/Producer.php                       |  4 +-
 src/Console/ConsumeCommand.php                |  4 +-
 src/Console/ReloadConsumersCommand.php        | 24 ++++-
 tests/Integration/ManagerTest.php             | 72 ++++++++++++++
 .../ReloadConsumersCommandTest.php            | 96 +++++++++++++++++++
 tests/Support/Helper/Producer.php             |  6 +-
 12 files changed, 242 insertions(+), 22 deletions(-)
 create mode 100644 tests/Integration/ManagerTest.php
 create mode 100644 tests/Integration/ReloadConsumersCommandTest.php

diff --git a/CLAUDE.md b/CLAUDE.md
index 966a449..d45e1cb 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -88,15 +88,15 @@ Joby sdílející `serialGroup` běží striktně v pořadí podle ID. `checkUnf
 
 ### Brokerová abstrakce (src/Broker/)
 
-`Producer` a `Consumer` jsou rozhraní - přines si libovolný broker. K dispozici je hotová implementace `PhpAmqpLib` (volitelná závislost; viz README pro doporučené omezení `conflict` pinující `php-amqplib` na `^3.0`). Priorita je modelována jako **samostatné fronty** pojmenované `<queue>_<priority>`; `QUEUE_TOP_PRIORITY = 0` je vyhrazena pro řídicí (DIE) zprávy, takže ji konzumeři kontrolují jako první. `publishDie()` + tělo `DIE` je způsob, jak `reload-consumers` elegantně restartuje běžící konzumery.
+`Producer` a `Consumer` jsou rozhraní - přines si libovolný broker. K dispozici je hotová implementace `PhpAmqpLib` (volitelná závislost; viz README pro doporučené omezení `conflict` pinující `php-amqplib` na `^3.0`). Priorita je modelována jako **samostatné fronty** pojmenované `<queue>_<priority>`; `QUEUE_TOP_PRIORITY = 0` je vyhrazena pro řídicí (DIE) zprávy, takže ji konzumeři kontrolují jako první. `publishDie()` + tělo `DIE` je způsob, jak `reload-consumers` elegantně restartuje běžící konzumery. Konzumer spuštěný s **labelem** (`consume -l <label>`) dostane vlastní top-priority frontu `<queue>_0_<label>` (`Manager::getTopPriorityName()` / `includeTopPriority()`); díky tomu lze přes `reload-consumers -l label1,label2` restartovat cíleně právě jeho a DIE zprávu nesní jiný konzumer. Label nesmí obsahovat oddělovač `_` (`QUEUE_NAME_PARTS_DELIMITER`).
 
 ### Konzolové příkazy (src/Console/)
 
 Všechny příkazy kromě `ConsumeCommand` rozšiřují lokální abstraktní `Command`, který obaluje běh do `ADT\CommandLock` (`FileSystemStorage` pod `tempDir`), takže nemohou běžet souběžně samy se sebou.
 
 - `background-queue:process` - vstupní bod pro cron (spouštět každou minutu).
-- `background-queue:consume [queue] -j <jobs> -p <priorities>` - dlouhoběžící brokerový konzumer; `-p` přijímá rozsahy jako `20-40`, `25-`, `-20`.
-- `background-queue:clear-finished [days]`, `background-queue:reload-consumers <number> [queue]`, `background-queue:update-schema`.
+- `background-queue:consume [queue] -j <jobs> -p <priorities> -l <label>` - dlouhoběžící brokerový konzumer; `-p` přijímá rozsahy jako `20-40`, `25-`, `-20`; `-l` je volitelný label pro cílený restart.
+- `background-queue:clear-finished [days]`, `background-queue:reload-consumers <number> [queue] [-l label1,label2]`, `background-queue:update-schema`.
 
 ### Hromadný insert
 
diff --git a/README.md b/README.md
index 9963d95..23a05ce 100644
--- a/README.md
+++ b/README.md
@@ -207,7 +207,7 @@ Ve všech ostatních případech se záznam uloží jako úspěšně dokončený
 
 `background-queue:clear-finished 14` Smaže všechny úspěšně zpracované záznamy starší 14 dní.
 
-`background-queue:reload-consumers QUEUE NUMBER` Reloadne NUMBER consumerů pro danou QUEUE.
+`background-queue:reload-consumers NUMBER [QUEUE] [-l LABEL1,LABEL2,...]` Pošle NUMBER restartovacích (DIE) zpráv. Bez `-l` do sdílené DIE fronty dané QUEUE, s `-l` cíleně do DIE fronty každého uvedeného labelu (viz `background-queue:consume -l`).
 
 `background-queue:update-schema` Aktualizuje databázové schéma, pokud je potřeba.
 
@@ -294,6 +294,9 @@ Můžeme tedy jednoho konzumera vyhradit například na rozesílání registrač
 Tím zajistíme, že rychlé odeslání registračního emailu nebude čekat na dlouho trvající úlohy, protože je odbaví první konzumer.
 Ale pokud by se vyskytlo více požadavků na zasílání emailů, po nějaké době je začnou odbavovat všichni konzumeři.
 
+Příkazu `background-queue:consume` máme dále možnost nastavit parametrem `-l` (label) jmenovku konzumera. Konzumer s labelem dostane vlastní DIE frontu (`<queue>_0_<label>`), takže ho lze při `background-queue:reload-consumers` restartovat cíleně pomocí `-l label1,label2,...`, aniž by DIE zprávy "snědl" jiný konzumer.
+Aby šel každý konzumer restartovat samostatně, dej každému unikátní label - konzumeři se stejným labelem totiž jednu DIE frontu sdílejí. Bez labelu zůstává chování jako dřív: všichni konzumeři sdílejí jednu DIE frontu.
+
 Dále máme možnost prioritu nastavenou pro callback přetížit při vkládání záznamu v metodě `publish`. Například víme, že se jedná o rozesílání newsletterů.
 Tedy se jedná o zasílání emailů, ale s nízkou prioritou zpracování.
 
diff --git a/src/Broker/Consumer.php b/src/Broker/Consumer.php
index 3229a39..d6a3231 100644
--- a/src/Broker/Consumer.php
+++ b/src/Broker/Consumer.php
@@ -4,5 +4,5 @@
 
 interface Consumer
 {
-	public function consume(string $queue, array $priorities): void;
+	public function consume(string $queue, array $priorities, ?string $consumerLabel = null): void;
 }
\ No newline at end of file
diff --git a/src/Broker/PhpAmqpLib/Consumer.php b/src/Broker/PhpAmqpLib/Consumer.php
index e81c293..c0edaa5 100644
--- a/src/Broker/PhpAmqpLib/Consumer.php
+++ b/src/Broker/PhpAmqpLib/Consumer.php
@@ -15,13 +15,14 @@ public function __construct(private Manager $manager, private BackgroundQueue $b
 	/**
 	 * @throws Exception
 	 */
-	public function consume(string $queue, array $priorities): void
+	public function consume(string $queue, array $priorities, ?string $consumerLabel = null): void
 	{
 		// TODO Do budoucna cheme podporovat libovolné priority a ne pouze jejich výčet.
 		//      Zde si musíme vytáhnout seznam existujících front. To lze přes HTTP API pomocí CURL.
 
 		// Nejprve se chceme kouknout, jestli není zaslána zpráva k ukončení, proto na první místo dáme TOP_PRIORITY frontu.
-		array_unshift($priorities, Manager::QUEUE_TOP_PRIORITY);
+		// S labelem dostane konzumer vlastní DIE frontu "0_<label>", takže ho lze restartovat cíleně.
+		$priorities = $this->manager->includeTopPriority($priorities, $consumerLabel);
 
 		// Sestavíme si seznam názvů front v RabbitMQ (tedy včetně priorit) a všechny inicializujeme
 		$queuesWithPriorities = [];
diff --git a/src/Broker/PhpAmqpLib/Manager.php b/src/Broker/PhpAmqpLib/Manager.php
index 620186e..eccd94c 100644
--- a/src/Broker/PhpAmqpLib/Manager.php
+++ b/src/Broker/PhpAmqpLib/Manager.php
@@ -10,6 +10,7 @@
 class Manager
 {
 	const QUEUE_TOP_PRIORITY = 0;
+	const QUEUE_NAME_PARTS_DELIMITER = '_';
 
 	private array $connectionParams;
 	private array $queueParams;
@@ -144,8 +145,37 @@ public function setupQos(): void
 		$this->initQos = true;
 	}
 
-	public function getQueueWithPriority(string $queue, int $priority): string
+	public function getQueueWithPriority(string $queue, string $priority): string
 	{
-		return $queue . '_' . $priority;
+		return $queue . self::QUEUE_NAME_PARTS_DELIMITER . $priority;
+	}
+
+	/**
+	 * Vloží na začátek seznamu priorit název top-priority (DIE) fronty.
+	 * Konzumer ji tak vždy kontroluje jako první - aby na DIE zprávu reagoval přednostně.
+	 */
+	public function includeTopPriority(array $priorities, ?string $label = null): array
+	{
+		array_unshift($priorities, $this->getTopPriorityName($label));
+		return $priorities;
+	}
+
+	/**
+	 * Vrátí název top-priority fronty. Bez labelu je to sdílená "0" fronta;
+	 * s labelem vznikne samostatná "0_<label>" fronta, díky níž má každý takto označený
+	 * konzumer vlastní DIE frontu a lze ho restartovat cíleně (viz ReloadConsumersCommand).
+	 */
+	public function getTopPriorityName(?string $label = null): string
+	{
+		$topPriority = (string) self::QUEUE_TOP_PRIORITY;
+		if (!is_null($label)) {
+			if (strpos($label, self::QUEUE_NAME_PARTS_DELIMITER) !== false) {
+				throw new Exception('Label cannot contain "' . self::QUEUE_NAME_PARTS_DELIMITER . '".');
+			}
+
+			$topPriority .= self::QUEUE_NAME_PARTS_DELIMITER . $label;
+		}
+
+		return $topPriority;
 	}
 }
\ No newline at end of file
diff --git a/src/Broker/PhpAmqpLib/Producer.php b/src/Broker/PhpAmqpLib/Producer.php
index 824f715..50a00d3 100644
--- a/src/Broker/PhpAmqpLib/Producer.php
+++ b/src/Broker/PhpAmqpLib/Producer.php
@@ -18,7 +18,7 @@ public function __construct(private Manager $manager)
 	/**
 	 * @throws Exception
 	 */
-	public function publish(string $id, string $queue, int $priority, ?int $expiration = null): void
+	public function publish(string $id, string $queue, string $priority, ?int $expiration = null): void
 	{
 		$queue = $this->manager->getQueueWithPriority($queue, $priority);
 		$exchange = $queue;
@@ -49,9 +49,9 @@ public function publish(string $id, string $queue, int $priority, ?int $expirati
 	/**
 	 * @throws Exception
 	 */
-	public function publishDie(string $queue): void
+	public function publishDie(string $queue, ?string $consumerLabel = null): void
 	{
-		$this->publish(self::DIE, $queue, Manager::QUEUE_TOP_PRIORITY);
+		$this->publish(self::DIE, $queue, $this->manager->getTopPriorityName($consumerLabel));
 	}
 
 	private function createMessage(string $body): AMQPMessage
diff --git a/src/Broker/Producer.php b/src/Broker/Producer.php
index a1cc19d..68d16f0 100644
--- a/src/Broker/Producer.php
+++ b/src/Broker/Producer.php
@@ -4,6 +4,6 @@
 
 interface Producer
 {
-	public function publish(string $id, string $queue, int $priority, ?int $expiration = null): void;
-	public function publishDie(string $queue): void;
+	public function publish(string $id, string $queue, string $priority, ?int $expiration = null): void;
+	public function publishDie(string $queue, ?string $consumerLabel = null): void;
 }
\ No newline at end of file
diff --git a/src/Console/ConsumeCommand.php b/src/Console/ConsumeCommand.php
index 263f57f..73ac43a 100644
--- a/src/Console/ConsumeCommand.php
+++ b/src/Console/ConsumeCommand.php
@@ -26,6 +26,7 @@ protected function configure(): void
 		$this->addArgument('queue', InputArgument::OPTIONAL);
 		$this->addOption('jobs', 'j', InputOption::VALUE_REQUIRED, 'Number of jobs consumed by one consumer in one process', 1);
 		$this->addOption('priorities', 'p', InputOption::VALUE_REQUIRED, 'Priorities for consume (e.g. 10, 20-40, 25-, -20)');
+		$this->addOption('label', 'l', InputOption::VALUE_OPTIONAL, 'Consumer label for targeted restart via reload-consumers command');
 	}
 
 	/**
@@ -35,6 +36,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 	{
 		$jobs = $input->getOption('jobs');
 		$priorities = $this->getPrioritiesListBasedConfig($input->getOption('priorities'));
+		$label = $input->getOption('label');
 
 		if (!is_numeric($jobs)) {
 			$output->writeln("<error>Option --jobs has to be integer</error>");
@@ -43,7 +45,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
 		for ($i = 0; $i < (int)$jobs; $i++) {
 			$this->backgroundQueue->dieIfNecessary();
-			$this->consumer->consume($this->backgroundQueue->getQueue($input->getArgument('queue')), $priorities);
+			$this->consumer->consume($this->backgroundQueue->getQueue($input->getArgument('queue')), $priorities, $label);
 		}
 
 		return self::SUCCESS;
diff --git a/src/Console/ReloadConsumersCommand.php b/src/Console/ReloadConsumersCommand.php
index 0eb405d..8ecba79 100644
--- a/src/Console/ReloadConsumersCommand.php
+++ b/src/Console/ReloadConsumersCommand.php
@@ -7,9 +7,10 @@
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 
-#[AsCommand(name: 'background-queue:reload-consumers', description: 'Creates the specified number of noop messages to reload consumers consuming specified queue.')]
+#[AsCommand(name: 'background-queue:reload-consumers', description: 'Restarts consumers by sending DIE messages to their (optionally label-specific) top-priority queue.')]
 class ReloadConsumersCommand extends Command
 {
 	public function __construct(private readonly BackgroundQueue $backgroundQueue, private readonly Producer $producer)
@@ -22,19 +23,34 @@ protected function configure(): void
 		$this->addArgument(
 			"number",
 			InputArgument::REQUIRED,
-			'Number of consumers to reload.'
+			'Number of DIE messages to send to each targeted consumer queue.'
 		);
 		$this->addArgument(
 			"queue",
 			InputArgument::OPTIONAL,
 			'A queue whose consumers are to reload.'
 		);
+		$this->addOption(
+			'label',
+			'l',
+			InputOption::VALUE_OPTIONAL,
+			'Comma-separated consumer labels to restart (see consume --label). Empty targets the shared DIE queue.'
+		);
 	}
 
 	protected function executeCommand(InputInterface $input, OutputInterface $output): int
 	{
-		for ($i = 0; $i < $input->getArgument("number"); $i++) {
-			$this->producer->publishDie($this->backgroundQueue->getQueue($input->getArgument("queue")));
+		// Labely cílí restart na konkrétní konzumery (každý má vlastní DIE frontu "0_<label>").
+		// Bez labelu se posílá do sdílené DIE fronty - zpětně kompatibilní původní chování.
+		$labels = $input->getOption('label');
+		$labels = $labels ? explode(',', $labels) : [null];
+
+		$queue = $this->backgroundQueue->getQueue($input->getArgument("queue"));
+
+		foreach ($labels as $label) {
+			for ($i = 0; $i < $input->getArgument("number"); $i++) {
+				$this->producer->publishDie($queue, $label);
+			}
 		}
 
 		return self::SUCCESS;
diff --git a/tests/Integration/ManagerTest.php b/tests/Integration/ManagerTest.php
new file mode 100644
index 0000000..d667a34
--- /dev/null
+++ b/tests/Integration/ManagerTest.php
@@ -0,0 +1,72 @@
+<?php
+
+namespace Tests\Integration;
+
+use ADT\BackgroundQueue\Broker\PhpAmqpLib\Manager;
+use Codeception\Test\Unit;
+use Tests\Support\IntegrationTester;
+
+/**
+ * Testuje pojmenování front v Manageru se zaměřením na label-specific DIE fronty
+ * (feature pro cílený restart konzumerů). Čistá logika - nepotřebuje DB ani RabbitMQ.
+ */
+class ManagerTest extends Unit
+{
+	protected IntegrationTester $tester;
+
+	private function getManager(): Manager
+	{
+		return new Manager([], ['arguments' => []]);
+	}
+
+	public function testGetTopPriorityNameWithoutLabel()
+	{
+		// Bez labelu zůstává sdílená top-priority (DIE) fronta "0".
+		$this->tester->assertSame('0', $this->getManager()->getTopPriorityName());
+		$this->tester->assertSame('0', $this->getManager()->getTopPriorityName(null));
+	}
+
+	public function testGetTopPriorityNameWithLabel()
+	{
+		// S labelem vznikne samostatná DIE fronta "0_<label>".
+		$this->tester->assertSame('0_consumer1', $this->getManager()->getTopPriorityName('consumer1'));
+	}
+
+	public function testGetTopPriorityNameRejectsLabelWithDelimiter()
+	{
+		// Label se vkládá do názvu fronty za oddělovač "_", takže ho sám obsahovat nesmí.
+		try {
+			$this->getManager()->getTopPriorityName('foo_bar');
+			$this->tester->fail('Očekávána výjimka pro label obsahující "_".');
+		} catch (\Exception $e) {
+			$this->tester->assertSame('Label cannot contain "_".', $e->getMessage());
+		}
+	}
+
+	public function testIncludeTopPriorityPrependsSharedQueue()
+	{
+		// Top-priority fronta se vkládá na začátek, aby ji konzumer kontroloval jako první.
+		$this->tester->assertSame(['0', 10, 20], $this->getManager()->includeTopPriority([10, 20]));
+	}
+
+	public function testIncludeTopPriorityPrependsLabelledQueue()
+	{
+		$this->tester->assertSame(['0_worker', 10, 20], $this->getManager()->includeTopPriority([10, 20], 'worker'));
+	}
+
+	public function testGetQueueWithPriority()
+	{
+		$this->tester->assertSame('general_10', $this->getManager()->getQueueWithPriority('general', '10'));
+	}
+
+	public function testGetQueueWithPrioritySupportsNamedQueueAndLabelledTopPriority()
+	{
+		// Regrese: pojmenovaná fronta (obsahuje "_") v kombinaci s label DIE prioritou ("0_label")
+		// musí projít. Dřívější kontrola na "_" v názvu fronty by tohle chybně odmítla.
+		$manager = $this->getManager();
+		$this->tester->assertSame(
+			'general_myqueue_0_worker',
+			$manager->getQueueWithPriority('general_myqueue', $manager->getTopPriorityName('worker'))
+		);
+	}
+}
diff --git a/tests/Integration/ReloadConsumersCommandTest.php b/tests/Integration/ReloadConsumersCommandTest.php
new file mode 100644
index 0000000..f4eb595
--- /dev/null
+++ b/tests/Integration/ReloadConsumersCommandTest.php
@@ -0,0 +1,96 @@
+<?php
+
+namespace Tests\Integration;
+
+use ADT\BackgroundQueue\BackgroundQueue;
+use ADT\BackgroundQueue\Broker\Producer;
+use ADT\BackgroundQueue\Console\ReloadConsumersCommand;
+use Codeception\Test\Unit;
+use Doctrine\DBAL\DriverManager;
+use Symfony\Component\Console\Input\ArrayInput;
+use Symfony\Component\Console\Output\NullOutput;
+use Tests\Support\IntegrationTester;
+
+/**
+ * Ověřuje, že reload-consumers pošle DIE zprávy do správných (label-specific) front
+ * a ve správném počtu - tedy jádro cíleného restartu konzumerů.
+ */
+class ReloadConsumersCommandTest extends Unit
+{
+	protected IntegrationTester $tester;
+
+	/**
+	 * Spustí executeCommand() přímo (obejde zámek z abstraktního Command) a vrátí
+	 * seznam volání publishDie() ve tvaru [['queue' => ..., 'label' => ...], ...].
+	 */
+	private function runReload(array $input): array
+	{
+		$producer = new class implements Producer {
+			public array $dieCalls = [];
+			public function publish(string $id, string $queue, string $priority, ?int $expiration = null): void {}
+			public function publishDie(string $queue, ?string $consumerLabel = null): void
+			{
+				$this->dieCalls[] = ['queue' => $queue, 'label' => $consumerLabel];
+			}
+		};
+
+		$backgroundQueue = new BackgroundQueue([
+			'queue' => 'general',
+			'priorities' => [10],
+			'connection' => DriverManager::getConnection(BackgroundQueue::parseDsn(self::getDsn())),
+			'logger' => null,
+			'producer' => $producer,
+		]);
+
+		$command = new ReloadConsumersCommand($backgroundQueue, $producer);
+
+		$arrayInput = new ArrayInput($input);
+		$arrayInput->bind($command->getDefinition());
+
+		$method = (new \ReflectionObject($command))->getMethod('executeCommand');
+		$method->setAccessible(true);
+		$method->invoke($command, $arrayInput, new NullOutput());
+
+		return $producer->dieCalls;
+	}
+
+	public function testWithoutLabelSendsToSharedQueue()
+	{
+		// Bez labelu se posílá NUMBER DIE zpráv do sdílené DIE fronty (label = null) - původní chování.
+		$calls = $this->runReload(['number' => 3]);
+
+		$this->tester->assertCount(3, $calls);
+		foreach ($calls as $call) {
+			$this->tester->assertSame('general', $call['queue']);
+			$this->tester->assertNull($call['label']);
+		}
+	}
+
+	public function testWithLabelsSendsNumberPerLabel()
+	{
+		// NUMBER zpráv na každý vyjmenovaný label - cílený restart konkrétních konzumerů.
+		$calls = $this->runReload(['number' => 2, '--label' => 'a,b']);
+
+		$this->tester->assertSame([
+			['queue' => 'general', 'label' => 'a'],
+			['queue' => 'general', 'label' => 'a'],
+			['queue' => 'general', 'label' => 'b'],
+			['queue' => 'general', 'label' => 'b'],
+		], $calls);
+	}
+
+	public function testRespectsNamedQueue()
+	{
+		// Volitelný argument queue rozšíří základní frontu (general -> general_myqueue).
+		$calls = $this->runReload(['number' => 1, 'queue' => 'myqueue', '--label' => 'x']);
+
+		$this->tester->assertSame([
+			['queue' => 'general_myqueue', 'label' => 'x'],
+		], $calls);
+	}
+
+	private static function getDsn(): string
+	{
+		return 'mysql://' . $_ENV['PROJECT_DB_USER'] . ':' . $_ENV['PROJECT_DB_PASSWORD'] . '@' . $_ENV['PROJECT_DB_HOST'] . ':' . $_ENV['PROJECT_DB_PORT'] . '/' . $_ENV['PROJECT_DB_DBNAME'];
+	}
+}
diff --git a/tests/Support/Helper/Producer.php b/tests/Support/Helper/Producer.php
index 20bae8c..14b82a2 100644
--- a/tests/Support/Helper/Producer.php
+++ b/tests/Support/Helper/Producer.php
@@ -22,7 +22,7 @@ class Producer implements \ADT\BackgroundQueue\Broker\Producer
 	 * (livelock). V produkci k tomu nedochází právě díky tomuto zpoždění, ne přes skutečné AMQP TTL fronty
 	 * (ty by mezi testy protékaly a nafukovaly getMessageCount).
 	 *
-	 * @var array<int, array{queue: string, priority: int, id: string}>
+	 * @var array<int, array{queue: string, priority: string, id: string}>
 	 */
 	private array $delayedMessages = [];
 
@@ -31,7 +31,7 @@ class Producer implements \ADT\BackgroundQueue\Broker\Producer
 	 * (viz reálný Producer / Manager::getQueueWithPriority). Helper to napodobuje,
 	 * aby šel otestovat prioritní routing i konzumace v pořadí priorit.
 	 */
-	public function publish(string $id, string $queue, int $priority, ?int $expiration = null): void
+	public function publish(string $id, string $queue, string $priority, ?int $expiration = null): void
 	{
 		// Zpožděné doručení (postponeBy): nevkládáme do prioritní fronty hned, ale do zpožděného bufferu,
 		// odkud se zpráva vydá až když nejsou žádné okamžité zprávy (viz $delayedMessages a consume()).
@@ -47,7 +47,7 @@ public function publish(string $id, string $queue, int $priority, ?int $expirati
 		$this->getChannel()->wait_for_pending_acks();
 	}
 
-	public function publishDie(string $queue): void
+	public function publishDie(string $queue, ?string $consumerLabel = null): void
 	{
 
 	}

From 293a6c71e7a49d6448f1816eaea9b57881333c84 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Viktor=20Ma=C5=A1=C3=AD=C4=8Dek?= <viktor@appsdevteam.com>
Date: Tue, 23 Jun 2026 12:51:28 +0200
Subject: [PATCH 2/2] Implement nice shutdown for graceful consumer termination

- Add SHUTDOWN control message and NICE_SHUTDOWN_EXIT_CODE (100) to PhpAmqpLib producer
- Implement publishShutdown() method on Producer interface for targeted graceful consumer stops
- Update PhpAmqpLib consumer to recognize SHUTDOWN message and exit with code 100 instead of restarting
- Add background-queue:shutdown-consumers command with same targeting as reload-consumers (number, queue, -l label)
- Add comprehensive test suite including end-to-end subprocess test verifying proper exit code
- Add shutdownConsumerWorker.php fixture for subprocess testing against real RabbitMQ and database
- Document nice shutdown mechanism in README with supervisor configuration example (exitcodes=100, autorestart=unexpected)
- Update CLAUDE.md with details on dual control messages (DIE for restart, SHUTDOWN for graceful stop)
- Update all Producer implementations and test mocks to support publishShutdown method

This allows supervisor-managed consumers to be gracefully stopped before server maintenance or restarts
without requiring restarts, while maintaining backward compatibility with existing reload-consumers workflow.
---
 CLAUDE.md                                     |   4 +-
 README.md                                     |  27 +++
 src/Broker/PhpAmqpLib/Consumer.php            |   7 +
 src/Broker/PhpAmqpLib/Producer.php            |  19 ++
 src/Broker/Producer.php                       |   1 +
 src/Console/ShutdownConsumersCommand.php      |  58 ++++++
 .../ReloadConsumersCommandTest.php            |   1 +
 .../ShutdownConsumersCommandTest.php          | 175 ++++++++++++++++++
 tests/Support/Data/shutdownConsumerWorker.php |  46 +++++
 tests/Support/Helper/Producer.php             |   5 +
 10 files changed, 342 insertions(+), 1 deletion(-)
 create mode 100644 src/Console/ShutdownConsumersCommand.php
 create mode 100644 tests/Integration/ShutdownConsumersCommandTest.php
 create mode 100644 tests/Support/Data/shutdownConsumerWorker.php

diff --git a/CLAUDE.md b/CLAUDE.md
index d45e1cb..e31c0f3 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -90,13 +90,15 @@ Joby sdílející `serialGroup` běží striktně v pořadí podle ID. `checkUnf
 
 `Producer` a `Consumer` jsou rozhraní - přines si libovolný broker. K dispozici je hotová implementace `PhpAmqpLib` (volitelná závislost; viz README pro doporučené omezení `conflict` pinující `php-amqplib` na `^3.0`). Priorita je modelována jako **samostatné fronty** pojmenované `<queue>_<priority>`; `QUEUE_TOP_PRIORITY = 0` je vyhrazena pro řídicí (DIE) zprávy, takže ji konzumeři kontrolují jako první. `publishDie()` + tělo `DIE` je způsob, jak `reload-consumers` elegantně restartuje běžící konzumery. Konzumer spuštěný s **labelem** (`consume -l <label>`) dostane vlastní top-priority frontu `<queue>_0_<label>` (`Manager::getTopPriorityName()` / `includeTopPriority()`); díky tomu lze přes `reload-consumers -l label1,label2` restartovat cíleně právě jeho a DIE zprávu nesní jiný konzumer. Label nesmí obsahovat oddělovač `_` (`QUEUE_NAME_PARTS_DELIMITER`).
 
+Vedle `DIE` existuje druhá řídicí zpráva `SHUTDOWN` (`publishShutdown()`, posílá ji `shutdown-consumers` se stejným cílením přes label). Liší se jen exit kódem po dojetí rozdělaného jobu: `DIE` ukončí konzumera přes `die()` (exit 0, supervisor ho typicky restartuje), `SHUTDOWN` přes `exit(Producer::NICE_SHUTDOWN_EXIT_CODE)` (= 100). Ten kód patří do supervisor `exitcodes` při `autorestart=unexpected`, takže konzumer už znovu nenaběhne - "nice shutdown" pro řízené zastavení (např. před restartem serveru). Detaily a vzor konfigurace supervisoru viz README sekce 6.
+
 ### Konzolové příkazy (src/Console/)
 
 Všechny příkazy kromě `ConsumeCommand` rozšiřují lokální abstraktní `Command`, který obaluje běh do `ADT\CommandLock` (`FileSystemStorage` pod `tempDir`), takže nemohou běžet souběžně samy se sebou.
 
 - `background-queue:process` - vstupní bod pro cron (spouštět každou minutu).
 - `background-queue:consume [queue] -j <jobs> -p <priorities> -l <label>` - dlouhoběžící brokerový konzumer; `-p` přijímá rozsahy jako `20-40`, `25-`, `-20`; `-l` je volitelný label pro cílený restart.
-- `background-queue:clear-finished [days]`, `background-queue:reload-consumers <number> [queue] [-l label1,label2]`, `background-queue:update-schema`.
+- `background-queue:clear-finished [days]`, `background-queue:reload-consumers <number> [queue] [-l label1,label2]`, `background-queue:shutdown-consumers <number> [queue] [-l label1,label2]` (řízené zastavení = reload, ale s exit kódem `NICE_SHUTDOWN_EXIT_CODE`, který supervisor nerestartuje), `background-queue:update-schema`.
 
 ### Hromadný insert
 
diff --git a/README.md b/README.md
index 23a05ce..c626ee5 100644
--- a/README.md
+++ b/README.md
@@ -209,6 +209,8 @@ Ve všech ostatních případech se záznam uloží jako úspěšně dokončený
 
 `background-queue:reload-consumers NUMBER [QUEUE] [-l LABEL1,LABEL2,...]` Pošle NUMBER restartovacích (DIE) zpráv. Bez `-l` do sdílené DIE fronty dané QUEUE, s `-l` cíleně do DIE fronty každého uvedeného labelu (viz `background-queue:consume -l`).
 
+`background-queue:shutdown-consumers NUMBER [QUEUE] [-l LABEL1,LABEL2,...]` Funguje stejně jako `reload-consumers` (stejné cílení přes QUEUE a `-l`), ale místo restartu konzumery **řízeně zastaví** - dojedou rozdělaný job, další si nevezmou a ukončí se exit kódem určeným k tomu, aby je supervisor už znovu nenastartoval (viz sekce [Řízené zastavení konzumerů](#6-řízené-zastavení-konzumerů-nice-shutdown)).
+
 `background-queue:update-schema` Aktualizuje databázové schéma, pokud je potřeba.
 
 Všechny commandy jsou chráněny proti vícenásobnému spuštění.
@@ -297,6 +299,31 @@ Ale pokud by se vyskytlo více požadavků na zasílání emailů, po nějaké d
 Příkazu `background-queue:consume` máme dále možnost nastavit parametrem `-l` (label) jmenovku konzumera. Konzumer s labelem dostane vlastní DIE frontu (`<queue>_0_<label>`), takže ho lze při `background-queue:reload-consumers` restartovat cíleně pomocí `-l label1,label2,...`, aniž by DIE zprávy "snědl" jiný konzumer.
 Aby šel každý konzumer restartovat samostatně, dej každému unikátní label - konzumeři se stejným labelem totiž jednu DIE frontu sdílejí. Bez labelu zůstává chování jako dřív: všichni konzumeři sdílejí jednu DIE frontu.
 
+### 6 Řízené zastavení konzumerů (nice shutdown)
+
+Konzumery typicky spouští a hlídá [supervisor](http://supervisord.org/) - když proces skončí, podle konfigurace ho znovu nastartuje. Toho využívá `background-queue:reload-consumers`: pošle DIE zprávu, konzumer se po dojetí rozdělaného jobu ukončí (exit kód `0`) a supervisor ho nahodí znovu. Tím se konzumeři "obmění" (např. kvůli nasazení nové verze kódu).
+
+Někdy ale chceme konzumery zastavit a **nenechat je znovu nastartovat** - třeba před restartem serveru nebo údržbou. K tomu slouží `background-queue:shutdown-consumers`. Funguje úplně stejně jako `reload-consumers` (stejné cílení přes QUEUE a `-l`), jen konzumer po dojetí rozdělaného jobu skončí dohodnutým **exit kódem `100`** (`ADT\BackgroundQueue\Broker\PhpAmqpLib\Producer::NICE_SHUTDOWN_EXIT_CODE`). V obou případech konzumer nejprve dokončí právě zpracovávaný job a teprve pak se ukončí - žádný job se neztratí ani nepřeruší.
+
+Aby supervisor uměl odlišit "restartuj" (reload) od "nech být" (shutdown), nastav v konfiguraci hlídaného programu `exitcodes` na shutdown exit kód a `autorestart=unexpected`:
+
+```ini
+[program:bq-consumer]
+command=php bin/console background-queue:consume
+autorestart=unexpected
+exitcodes=100
+```
+
+Sémantika `autorestart=unexpected` (mimochodem výchozí hodnota): supervisor restartuje proces jen tehdy, když skončí exit kódem, který **není** v `exitcodes`. Pak platí:
+
+| Exit kód | Situace | Je v `exitcodes=100`? | Chování supervisoru |
+|---|---|---|---|
+| `0` | reload (DIE) i běžný konec | ne | "unexpected" -> **restartuje** |
+| `100` | nice shutdown | ano (expected) | **nenastartuje** |
+| jiný | pád procesu | ne | "unexpected" -> restartuje |
+
+Výchozí `exitcodes` je `0`, proto je nutné ho přepsat na `100` - jinak by byl naopak restartován běžný konec a nice shutdown by se choval jako pád. Pozor také, že `autorestart` se uplatní až pro proces, který úspěšně naběhl (stav `RUNNING`); rozběh řídí `startsecs`/`startretries`.
+
 Dále máme možnost prioritu nastavenou pro callback přetížit při vkládání záznamu v metodě `publish`. Například víme, že se jedná o rozesílání newsletterů.
 Tedy se jedná o zasílání emailů, ale s nízkou prioritou zpracování.
 
diff --git a/src/Broker/PhpAmqpLib/Consumer.php b/src/Broker/PhpAmqpLib/Consumer.php
index c0edaa5..e598aa6 100644
--- a/src/Broker/PhpAmqpLib/Consumer.php
+++ b/src/Broker/PhpAmqpLib/Consumer.php
@@ -48,9 +48,16 @@ public function consume(string $queue, array $priorities, ?string $consumerLabel
 				$this->manager->closeChannel();
 
 				if ($msg->getBody() === Producer::DIE) {
+					// Restart: exit 0 -> supervisor konzumera (typicky) znovu nastartuje.
 					die();
 				}
 
+				if ($msg->getBody() === Producer::SHUTDOWN) {
+					// Nice shutdown: rozdělaný job je už hotový (zpracovává se sériově, prefetch 1), další si nebereme
+					// a ukončíme se dohodnutým exit kódem, který má supervisor v "exitcodes" - proces už nenaběhne.
+					exit(Producer::NICE_SHUTDOWN_EXIT_CODE);
+				}
+
 				$this->backgroundQueue->processJob((int)$msg->getBody());
 			});
 		}
diff --git a/src/Broker/PhpAmqpLib/Producer.php b/src/Broker/PhpAmqpLib/Producer.php
index 50a00d3..0559ba9 100644
--- a/src/Broker/PhpAmqpLib/Producer.php
+++ b/src/Broker/PhpAmqpLib/Producer.php
@@ -11,6 +11,14 @@
 {
 	const DIE = 'die';
 
+	// "Nice shutdown": konzumer dojede rozdělaný job, vezme si tuto řídicí zprávu místo dalšího jobu
+	// a ukončí se s NICE_SHUTDOWN_EXIT_CODE. Na rozdíl od DIE (exit 0, supervisor konzumera restartuje)
+	// je tento exit kód určen k zařazení do supervisor "exitcodes", takže proces už znovu nenaběhne.
+	// Slouží k řízenému zastavení konzumerů (např. před restartem serveru).
+	const SHUTDOWN = 'shutdown';
+
+	const NICE_SHUTDOWN_EXIT_CODE = 100;
+
 	public function __construct(private Manager $manager)
 	{
 	}
@@ -54,6 +62,17 @@ public function publishDie(string $queue, ?string $consumerLabel = null): void
 		$this->publish(self::DIE, $queue, $this->manager->getTopPriorityName($consumerLabel));
 	}
 
+	/**
+	 * Pošle do (případně label-specifické) DIE fronty zprávu pro "nice shutdown" - konzumer se po dojetí
+	 * rozdělaného jobu ukončí s NICE_SHUTDOWN_EXIT_CODE a supervisor ho už nenastartuje (viz README).
+	 *
+	 * @throws Exception
+	 */
+	public function publishShutdown(string $queue, ?string $consumerLabel = null): void
+	{
+		$this->publish(self::SHUTDOWN, $queue, $this->manager->getTopPriorityName($consumerLabel));
+	}
+
 	private function createMessage(string $body): AMQPMessage
 	{
 		$properties = [
diff --git a/src/Broker/Producer.php b/src/Broker/Producer.php
index 68d16f0..4bc308c 100644
--- a/src/Broker/Producer.php
+++ b/src/Broker/Producer.php
@@ -6,4 +6,5 @@ interface Producer
 {
 	public function publish(string $id, string $queue, string $priority, ?int $expiration = null): void;
 	public function publishDie(string $queue, ?string $consumerLabel = null): void;
+	public function publishShutdown(string $queue, ?string $consumerLabel = null): void;
 }
\ No newline at end of file
diff --git a/src/Console/ShutdownConsumersCommand.php b/src/Console/ShutdownConsumersCommand.php
new file mode 100644
index 0000000..9a40b68
--- /dev/null
+++ b/src/Console/ShutdownConsumersCommand.php
@@ -0,0 +1,58 @@
+<?php
+
+namespace ADT\BackgroundQueue\Console;
+
+use ADT\BackgroundQueue\BackgroundQueue;
+use ADT\BackgroundQueue\Broker\Producer;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+
+#[AsCommand(name: 'background-queue:shutdown-consumers', description: 'Gracefully stops consumers by sending shutdown messages to their (optionally label-specific) top-priority queue. Unlike reload-consumers, consumers exit with a code meant to keep the supervisor from restarting them.')]
+class ShutdownConsumersCommand extends Command
+{
+	public function __construct(private readonly BackgroundQueue $backgroundQueue, private readonly Producer $producer)
+	{
+		parent::__construct();
+	}
+
+	protected function configure(): void
+	{
+		$this->addArgument(
+			"number",
+			InputArgument::REQUIRED,
+			'Number of shutdown messages to send to each targeted consumer queue.'
+		);
+		$this->addArgument(
+			"queue",
+			InputArgument::OPTIONAL,
+			'A queue whose consumers are to shut down.'
+		);
+		$this->addOption(
+			'label',
+			'l',
+			InputOption::VALUE_OPTIONAL,
+			'Comma-separated consumer labels to shut down (see consume --label). Empty targets the shared DIE queue.'
+		);
+	}
+
+	protected function executeCommand(InputInterface $input, OutputInterface $output): int
+	{
+		// Stejné cílení jako reload-consumers (sdílená vs. label-specifická DIE fronta), liší se jen typem
+		// řídicí zprávy: shutdown konzumera ukončí tak, aby ho supervisor už znovu nenastartoval (viz README).
+		$labels = $input->getOption('label');
+		$labels = $labels ? explode(',', $labels) : [null];
+
+		$queue = $this->backgroundQueue->getQueue($input->getArgument("queue"));
+
+		foreach ($labels as $label) {
+			for ($i = 0; $i < $input->getArgument("number"); $i++) {
+				$this->producer->publishShutdown($queue, $label);
+			}
+		}
+
+		return self::SUCCESS;
+	}
+}
diff --git a/tests/Integration/ReloadConsumersCommandTest.php b/tests/Integration/ReloadConsumersCommandTest.php
index f4eb595..092feb5 100644
--- a/tests/Integration/ReloadConsumersCommandTest.php
+++ b/tests/Integration/ReloadConsumersCommandTest.php
@@ -32,6 +32,7 @@ public function publishDie(string $queue, ?string $consumerLabel = null): void
 			{
 				$this->dieCalls[] = ['queue' => $queue, 'label' => $consumerLabel];
 			}
+			public function publishShutdown(string $queue, ?string $consumerLabel = null): void {}
 		};
 
 		$backgroundQueue = new BackgroundQueue([
diff --git a/tests/Integration/ShutdownConsumersCommandTest.php b/tests/Integration/ShutdownConsumersCommandTest.php
new file mode 100644
index 0000000..3f3b35f
--- /dev/null
+++ b/tests/Integration/ShutdownConsumersCommandTest.php
@@ -0,0 +1,175 @@
+<?php
+
+namespace Tests\Integration;
+
+use ADT\BackgroundQueue\BackgroundQueue;
+use ADT\BackgroundQueue\Broker\PhpAmqpLib\Manager;
+use ADT\BackgroundQueue\Broker\PhpAmqpLib\Producer as AmqpProducer;
+use ADT\BackgroundQueue\Broker\Producer;
+use ADT\BackgroundQueue\Console\ShutdownConsumersCommand;
+use Codeception\Test\Unit;
+use Doctrine\DBAL\DriverManager;
+use Symfony\Component\Console\Input\ArrayInput;
+use Symfony\Component\Console\Output\NullOutput;
+use Tests\Support\IntegrationTester;
+
+/**
+ * Ověřuje, že shutdown-consumers pošle "nice shutdown" zprávy do správných (label-specific) front
+ * a ve správném počtu - tedy jádro cíleného řízeného zastavení konzumerů (analogie reload-consumers).
+ */
+class ShutdownConsumersCommandTest extends Unit
+{
+	protected IntegrationTester $tester;
+
+	/**
+	 * Spustí executeCommand() přímo (obejde zámek z abstraktního Command) a vrátí
+	 * seznam volání publishShutdown() ve tvaru [['queue' => ..., 'label' => ...], ...].
+	 */
+	private function runShutdown(array $input): array
+	{
+		$producer = new class implements Producer {
+			public array $shutdownCalls = [];
+			public function publish(string $id, string $queue, string $priority, ?int $expiration = null): void {}
+			public function publishDie(string $queue, ?string $consumerLabel = null): void {}
+			public function publishShutdown(string $queue, ?string $consumerLabel = null): void
+			{
+				$this->shutdownCalls[] = ['queue' => $queue, 'label' => $consumerLabel];
+			}
+		};
+
+		$backgroundQueue = new BackgroundQueue([
+			'queue' => 'general',
+			'priorities' => [10],
+			'connection' => DriverManager::getConnection(BackgroundQueue::parseDsn(self::getDsn())),
+			'logger' => null,
+			'producer' => $producer,
+		]);
+
+		$command = new ShutdownConsumersCommand($backgroundQueue, $producer);
+
+		$arrayInput = new ArrayInput($input);
+		$arrayInput->bind($command->getDefinition());
+
+		$method = (new \ReflectionObject($command))->getMethod('executeCommand');
+		$method->setAccessible(true);
+		$method->invoke($command, $arrayInput, new NullOutput());
+
+		return $producer->shutdownCalls;
+	}
+
+	public function testWithoutLabelSendsToSharedQueue()
+	{
+		// Bez labelu se posílá NUMBER shutdown zpráv do sdílené DIE fronty (label = null) - stejné cílení jako reload.
+		$calls = $this->runShutdown(['number' => 3]);
+
+		$this->tester->assertCount(3, $calls);
+		foreach ($calls as $call) {
+			$this->tester->assertSame('general', $call['queue']);
+			$this->tester->assertNull($call['label']);
+		}
+	}
+
+	public function testWithLabelsSendsNumberPerLabel()
+	{
+		// NUMBER zpráv na každý vyjmenovaný label - cílené zastavení konkrétních konzumerů.
+		$calls = $this->runShutdown(['number' => 2, '--label' => 'a,b']);
+
+		$this->tester->assertSame([
+			['queue' => 'general', 'label' => 'a'],
+			['queue' => 'general', 'label' => 'a'],
+			['queue' => 'general', 'label' => 'b'],
+			['queue' => 'general', 'label' => 'b'],
+		], $calls);
+	}
+
+	public function testRespectsNamedQueue()
+	{
+		// Volitelný argument queue rozšíří základní frontu (general -> general_myqueue).
+		$calls = $this->runShutdown(['number' => 1, 'queue' => 'myqueue', '--label' => 'x']);
+
+		$this->tester->assertSame([
+			['queue' => 'general_myqueue', 'label' => 'x'],
+		], $calls);
+	}
+
+	/**
+	 * End-to-end ověření samotného exit kódu: reálný Producer pošle jednu SHUTDOWN zprávu do top-priority
+	 * (DIE) fronty, reálný konzumer (spuštěný jako samostatný proces, protože exit() by jinak zabil i test
+	 * runner) si pro ni po startu doběhne a ukončí se Producer::NICE_SHUTDOWN_EXIT_CODE.
+	 */
+	public function testConsumerExitsWithNiceShutdownCode()
+	{
+		$queue = 'shutdowntest';
+
+		$manager = new Manager(self::getRabbitParams(), ['arguments' => []]);
+
+		// Vyčistíme top-priority frontu od případných zbytků z dřívějška, ať si konzumer vezme právě naši SHUTDOWN.
+		$topPriorityQueue = $manager->getQueueWithPriority($queue, $manager->getTopPriorityName());
+		$manager->createExchange($topPriorityQueue);
+		$manager->createQueue($topPriorityQueue, $topPriorityQueue);
+		$manager->getChannel()->queue_purge($topPriorityQueue);
+
+		// Přesně jedna SHUTDOWN zpráva - leží ve frontě, než si pro ni konzumer po startu doběhne (žádný race).
+		(new AmqpProducer($manager))->publishShutdown($queue);
+		$manager->closeConnection();
+
+		$exitCode = $this->runWorker($queue, 15);
+
+		$this->tester->assertSame(AmqpProducer::NICE_SHUTDOWN_EXIT_CODE, $exitCode);
+	}
+
+	/**
+	 * Spustí konzumera (fixtura shutdownConsumerWorker.php) jako samostatný proces a vrátí jeho exit kód.
+	 * Hlídá timeout, aby test nezamrzl, kdyby konzumer SHUTDOWN nedostal a běžel dál v consume smyčce.
+	 */
+	private function runWorker(string $queue, int $timeoutSeconds): int
+	{
+		$worker = codecept_data_dir('shutdownConsumerWorker.php');
+
+		$descriptors = [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']];
+		$process = proc_open(['php', $worker, $queue], $descriptors, $pipes, null, $_ENV);
+		$this->tester->assertIsResource($process, 'Konzumera se nepodařilo spustit.');
+
+		$start = time();
+		do {
+			$status = proc_get_status($process);
+			if (!$status['running']) {
+				break;
+			}
+			usleep(100000);
+		} while (time() - $start < $timeoutSeconds);
+
+		// proc_get_status vrací platný exitcode jen při prvním volání po skončení procesu - přečteme ho teď.
+		$exitCode = $status['exitcode'];
+		$timedOut = $status['running'];
+
+		foreach ($pipes as $pipe) {
+			fclose($pipe);
+		}
+		if ($timedOut) {
+			proc_terminate($process, 9);
+		}
+		proc_close($process);
+
+		if ($timedOut) {
+			$this->tester->fail("Konzumer se neukončil do {$timeoutSeconds}s - SHUTDOWN nejspíš nedorazil.");
+		}
+
+		return $exitCode;
+	}
+
+	private static function getRabbitParams(): array
+	{
+		return [
+			'host' => $_ENV['PROJECT_RABBITMQ_HOST'],
+			'port' => $_ENV['PROJECT_RABBITMQ_PORT'],
+			'user' => $_ENV['PROJECT_RABBITMQ_USER'],
+			'password' => $_ENV['PROJECT_RABBITMQ_PASSWORD'],
+		];
+	}
+
+	private static function getDsn(): string
+	{
+		return 'mysql://' . $_ENV['PROJECT_DB_USER'] . ':' . $_ENV['PROJECT_DB_PASSWORD'] . '@' . $_ENV['PROJECT_DB_HOST'] . ':' . $_ENV['PROJECT_DB_PORT'] . '/' . $_ENV['PROJECT_DB_DBNAME'];
+	}
+}
diff --git a/tests/Support/Data/shutdownConsumerWorker.php b/tests/Support/Data/shutdownConsumerWorker.php
new file mode 100644
index 0000000..e774647
--- /dev/null
+++ b/tests/Support/Data/shutdownConsumerWorker.php
@@ -0,0 +1,46 @@
+<?php
+
+/**
+ * Fixtura pro ShutdownConsumersCommandTest::testConsumerExitsWithNiceShutdownCode.
+ *
+ * Spustí reálného PhpAmqpLib konzumera proti RabbitMQ. Jakmile z top-priority (DIE) fronty dostane
+ * SHUTDOWN zprávu, konzumer se ukončí kódem Producer::NICE_SHUTDOWN_EXIT_CODE. Skript běží jako
+ * samostatný proces právě proto, že exit() v konzumeru by jinak ukončil i celý test runner -
+ * jedině subprocess umožní exit kód odchytit a ověřit.
+ *
+ * Argument: $argv[1] = název základní fronty (musí sedět s frontou, do které test pošle SHUTDOWN).
+ */
+
+require dirname(__DIR__, 3) . '/vendor/autoload.php';
+
+use ADT\BackgroundQueue\BackgroundQueue;
+use ADT\BackgroundQueue\Broker\PhpAmqpLib\Consumer;
+use ADT\BackgroundQueue\Broker\PhpAmqpLib\Manager;
+use Doctrine\DBAL\DriverManager;
+
+$queue = $argv[1];
+
+$manager = new Manager(
+	[
+		'host' => $_ENV['PROJECT_RABBITMQ_HOST'],
+		'port' => $_ENV['PROJECT_RABBITMQ_PORT'],
+		'user' => $_ENV['PROJECT_RABBITMQ_USER'],
+		'password' => $_ENV['PROJECT_RABBITMQ_PASSWORD'],
+	],
+	['arguments' => []]
+);
+
+$dsn = 'mysql://' . $_ENV['PROJECT_DB_USER'] . ':' . $_ENV['PROJECT_DB_PASSWORD'] . '@' . $_ENV['PROJECT_DB_HOST'] . ':' . $_ENV['PROJECT_DB_PORT'] . '/' . $_ENV['PROJECT_DB_DBNAME'];
+
+$backgroundQueue = new BackgroundQueue([
+	'queue' => $queue,
+	'priorities' => [10],
+	'connection' => DriverManager::getConnection(BackgroundQueue::parseDsn($dsn)),
+	'logger' => null,
+	'producer' => null,
+]);
+
+$consumer = new Consumer($manager, $backgroundQueue);
+
+// Nekonečná consume smyčka; SHUTDOWN zpráva (čekající ve frontě) ji ukončí přes exit(NICE_SHUTDOWN_EXIT_CODE).
+$consumer->consume($queue, [10]);
diff --git a/tests/Support/Helper/Producer.php b/tests/Support/Helper/Producer.php
index 14b82a2..b0694f4 100644
--- a/tests/Support/Helper/Producer.php
+++ b/tests/Support/Helper/Producer.php
@@ -52,6 +52,11 @@ public function publishDie(string $queue, ?string $consumerLabel = null): void
 
 	}
 
+	public function publishShutdown(string $queue, ?string $consumerLabel = null): void
+	{
+
+	}
+
 	public function publishNoop(): void
 	{