[10.2.2] Possibility to declare quorum queue. (vyuldashev#359)

* Added: possibility to declare quorum queue using connection config or artisan command

* Ignore priority argument for quorum queue.

* Fixed style

Co-authored-by: Merhnferr <i.gaevskiy@andersenlab.com>

Author: Merhnferr

src/Console/QueueDeclareCommand.php

@@ -12,7 +12,8 @@ class QueueDeclareCommand extends Command
                            {name : The name of the queue to declare}
                            {connection=rabbitmq : The name of the queue connection to use}
                            {--durable=1}
-                           {--auto-delete=0}';
+                           {--auto-delete=0}
+                           {--quorum=0}';
 
     protected $description = 'Declare queue';
 
@@ -32,10 +33,15 @@ public function handle(RabbitMQConnector $connector): void
             return;
         }
 
+        $arguments = (bool) $this->option('quorum')
+            ? ['x-queue-type' => 'quorum']
+            : [];
+
         $queue->declareQueue(
             $this->argument('name'),
             (bool) $this->option('durable'),
-            (bool) $this->option('auto-delete')
+            (bool) $this->option('auto-delete'),
+            $arguments
         );
 
         $this->info('Queue declared successfully.');

src/Queue/RabbitMQQueue.php

@@ -611,7 +611,8 @@ protected function getQueueArguments(string $destination): array
         // Messages without a priority property are treated as if their priority were 0.
         // Messages with a priority which is higher than the queue's maximum, are treated as if they were
         // published with the maximum priority.
-        if ($this->isPrioritizeDelayed()) {
+        // Quorum queues does not support priority.
+        if ($this->isPrioritizeDelayed() && ! $this->isQuorum()) {
             $arguments['x-max-priority'] = $this->getQueueMaxPriority();
         }
 
@@ -620,6 +621,10 @@ protected function getQueueArguments(string $destination): array
             $arguments['x-dead-letter-routing-key'] = $this->getFailedRoutingKey($destination);
         }
 
+        if ($this->isQuorum()) {
+            $arguments['x-queue-type'] = 'quorum';
+        }
+
         return $arguments;
     }
 
@@ -710,6 +715,16 @@ protected function isRerouteFailed(): bool
         return (bool) (Arr::get($this->options, 'reroute_failed') ?: false);
     }
 
+    /**
+     * Returns &true;, if declared queue must be quorum queue.
+     *
+     * @return bool
+     */
+    protected function isQuorum(): bool
+    {
+        return (bool) (Arr::get($this->options, 'quorum') ?: false);
+    }
+
     /**
      * Get the exchange for failed messages.
      *

tests/Functional/RabbitMQQueueTest.php

@@ -139,6 +139,22 @@ public function testFailedRoutingKey(): void
         $this->assertSame('test.failed', $this->callMethod($queue, 'getFailedRoutingKey', ['test']));
     }
 
+    public function testQuorum(): void
+    {
+        /** @var $queue RabbitMQQueue */
+        $queue = $this->connection();
+        $this->assertFalse($this->callMethod($queue, 'isQuorum'));
+
+        $queue = $this->connection('rabbitmq-with-options');
+        $this->assertFalse($this->callMethod($queue, 'isQuorum'));
+
+        $queue = $this->connection('rabbitmq-with-quorum-options');
+        $this->assertTrue($this->callMethod($queue, 'isQuorum'));
+
+        $queue = $this->connection('rabbitmq-with-options-empty');
+        $this->assertFalse($this->callMethod($queue, 'isQuorum'));
+    }
+
     public function testDeclareDeleteExchange(): void
     {
         /** @var $queue RabbitMQQueue */
@@ -193,6 +209,17 @@ public function testQueueArguments(): void
         $this->assertEquals(array_keys($expected), array_keys($actual));
         $this->assertEquals(array_values($expected), array_values($actual));
 
+        $queue = $this->connection('rabbitmq-with-quorum-options');
+        $actual = $this->callMethod($queue, 'getQueueArguments', [$name]);
+        $expected = [
+            'x-dead-letter-exchange' => 'failed-exchange',
+            'x-dead-letter-routing-key' => sprintf('application-x.%s.failed', $name),
+            'x-queue-type' => 'quorum',
+        ];
+
+        $this->assertEquals(array_keys($expected), array_keys($actual));
+        $this->assertEquals(array_values($expected), array_values($actual));
+
         $queue = $this->connection('rabbitmq-with-options-empty');
         $actual = $this->callMethod($queue, 'getQueueArguments', [$name]);
         $expected = [];

tests/Functional/TestCase.php

@@ -110,6 +110,45 @@ protected function getEnvironmentSetUp($app): void
                     'reroute_failed' => '',
                     'failed_exchange' => '',
                     'failed_routing_key' => '',
+                    'quorum' => '',
+                ],
+            ],
+
+            'worker' => 'default',
+
+        ]);
+        $app['config']->set('queue.connections.rabbitmq-with-quorum-options', [
+            'driver' => 'rabbitmq',
+            'queue' => 'order',
+            'connection' => AMQPLazyConnection::class,
+
+            'hosts' => [
+                [
+                    'host' => getenv('HOST'),
+                    'port' => getenv('PORT'),
+                    'vhost' => '/',
+                    'user' => 'guest',
+                    'password' => 'guest',
+                ],
+            ],
+
+            'options' => [
+                'ssl_options' => [
+                    'cafile' => null,
+                    'local_cert' => null,
+                    'local_key' => null,
+                    'verify_peer' => true,
+                    'passphrase' => null,
+                ],
+
+                'queue' => [
+                    'exchange' => 'application-x',
+                    'exchange_type' => 'topic',
+                    'exchange_routing_key' => 'process.%s',
+                    'reroute_failed' => true,
+                    'failed_exchange' => 'failed-exchange',
+                    'failed_routing_key' => 'application-x.%s.failed',
+                    'quorum' => true,
                 ],
             ],