Use only single DLX queue

Author: bckp

src/Consumer/Consumer.php

@@ -83,6 +83,7 @@ protected function sendResponse(Message $message, Channel $channel, int $result,
 		match ($result) {
 			IConsumer::MESSAGE_ACK, IConsumer::MESSAGE_ACK_AND_TERMINATE => $channel->ack($message),
 			IConsumer::MESSAGE_NACK, IConsumer::MESSAGE_NACK_AND_TERMINATE => $channel->nack($message),
+			IConsumer::MESSAGE_NACK_REJECT, IConsumer::MESSAGE_NACK_REJECT_AND_TERMINATE => $channel->nack($message, requeue: false),
 			IConsumer::MESSAGE_REJECT, IConsumer::MESSAGE_REJECT_AND_TERMINATE => $channel->reject($message, false),
 			default => throw new \InvalidArgumentException("Unknown return value of consumer [{$this->name}] user callback"),
 		};
@@ -92,7 +93,8 @@ protected function sendResponse(Message $message, Channel $channel, int $result,
 			[
 				IConsumer::MESSAGE_REJECT_AND_TERMINATE,
 				IConsumer::MESSAGE_ACK_AND_TERMINATE,
-				IConsumer::MESSAGE_NACK_AND_TERMINATE
+				IConsumer::MESSAGE_NACK_AND_TERMINATE,
+				IConsumer::MESSAGE_NACK_REJECT_AND_TERMINATE,
 			],
 			true
 		)

src/Consumer/IConsumer.php

@@ -11,10 +11,12 @@ interface IConsumer
 
 	public const MESSAGE_ACK = 1;
 	public const MESSAGE_NACK = 2;
-	public const MESSAGE_REJECT = 3;
-	public const MESSAGE_REJECT_AND_TERMINATE = 4;
-	public const MESSAGE_ACK_AND_TERMINATE = 5;
-	public const MESSAGE_NACK_AND_TERMINATE = 6;
+	public const MESSAGE_NACK_REJECT = 3;
+	public const MESSAGE_REJECT = 4;
+	public const MESSAGE_ACK_AND_TERMINATE = 11;
+	public const MESSAGE_NACK_AND_TERMINATE = 12;
+	public const MESSAGE_NACK_REJECT_AND_TERMINATE = 13;
+	public const MESSAGE_REJECT_AND_TERMINATE = 14;
 
 	public function consume(Message $message): int;
 }

src/DI/Helpers/QueuesHelper.php

@@ -34,8 +34,8 @@ public function getQueueSchema(): Schema
 			'autoDelete' => Expect::bool(false),
 			'noWait' => Expect::bool(false),
 			'arguments' => Expect::array(),
-			'dlx' => Expect::arrayOf(Expect::int()->min(1))->required(false)->before(
-				fn(array $dlx): array => $this->normalizeDlx($dlx)
+			'dlx' => Expect::int()->min(1)->required(false)->before(
+				static fn(string $time): int => (int) strtotime($time, 0)
 			),
 			'autoCreate' => Expect::int(
 				AbstractDataBag::AutoCreateLazy
@@ -73,13 +73,4 @@ public function setup(ContainerBuilder $builder, array $config = []): ServiceDef
 			->setFactory(QueueFactory::class)
 			->setArguments([$queuesDataBag]);
 	}
-
-	/**
-	 * @param string[] $dlx
-	 * @return int[]
-	 */
-	protected function normalizeDlx(array $dlx): array
-	{
-		return array_map(static fn(string $time): int => (int) strtotime($time, 0), $dlx);
-	}
 }

src/DI/RabbitMQExtension.php

@@ -123,51 +123,42 @@ protected function processExtensions(array &$config): void
 				continue;
 			}
 
-			$exchangeRetry = $name . '.dlx-retry';
-			$exchangeDlx = $name . '.dlx-wait';
+			$exchangeOut = $name . '.dlx-out';
+			$exchangeIn = $name . '.dlx-in';
+
+			$queueDlx = sprintf('%s.dlx-%s', $name, (string) $data['dlx']);
+
+			# Setup dead letter exchange
+			$config['queues'][$name]['arguments']['x-dead-letter-exchange'] = $exchangeIn;
 
 			# DLX Exchange: will pass msg to queue
-			$config['exchanges'][$exchangeRetry] = $this->exchangesHelper->processConfiguration([
+			$config['exchanges'][$exchangeOut] = $this->exchangesHelper->processConfiguration([
 				'connection' => $data['connection'],
 				'type' => ExchangesHelper::ExchangeTypes[3],
 				'queueBindings' => [
-					$name => [
-						'routingKey' => [],
-					],
+					$name => [],
 				],
 			]);
 
 			# DLX Exchange: will pass msg to dlx queue
-			$exchangeDataBag = [
+			$config['exchanges'][$exchangeIn] = $this->exchangesHelper->processConfiguration([
 				'connection' => $data['connection'],
-				'type' => ExchangesHelper::ExchangeTypes[2], // headers
-				'queueBindings' => []
-			];
+				'type' => ExchangesHelper::ExchangeTypes[3],
+				'queueBindings' => [
+					$queueDlx => []
+				]
+			]);
 
 			# Expand dlx into new queues and exchange for them
-			foreach ($data['dlx'] as $pos => $seconds) {
-				$queueName = sprintf('%s.dlx-%s', $name, (string) $seconds);
-				$config['queues'][$queueName] = $this->queuesHelper->processConfiguration([
-					'connection' => $data['connection'],
-					'autoCreate' => true,
-					'arguments' => [
-						'x-dead-letter-exchange' => $exchangeRetry,
-						'x-message-ttl' => $seconds * 1000,
-					]
-				]);
-
-				$exchangeDataBag['queueBindings'][$queueName] = [
-					'arguments' => [
-						'x-match' => 'all',
-						'x-death' => [
-							'name' => $name,
-							'count' => $pos * 2,
-						],
-					],
-				];
-			}
+			$config['queues'][$queueDlx] = $this->queuesHelper->processConfiguration([
+				'connection' => $data['connection'],
+				'autoCreate' => true,
+				'arguments' => [
+					'x-dead-letter-exchange' => $exchangeOut,
+					'x-message-ttl' => $data['dlx'] * 1000,
+				]
+			]);
 
-			$config['exchanges'][$exchangeDlx] =  $this->exchangesHelper->processConfiguration($exchangeDataBag);
 			unset($config['queues'][$name]['dlx']);
 		}
 	}