resolve conflicts

Author: vyuldashev

src/Console/ConsumeCommand.php

@@ -24,7 +24,8 @@ class ConsumeCommand extends WorkCommand
                             {--timeout=60 : The number of seconds a child process can run}
                             {--tries=1 : Number of times to attempt a job before logging it failed}
                             {--rest=0 : Number of seconds to rest between jobs}
-                           
+
+                            {--max-priority=}
                             {--consumer-tag}
                             {--prefetch-size=0}
                             {--prefetch-count=1000}
@@ -40,6 +41,7 @@ public function handle(): void
         $consumer->setContainer($this->laravel);
         $consumer->setName($this->option('name'));
         $consumer->setConsumerTag($this->consumerTag());
+        $consumer->setMaxPriority((int) $this->option('max-priority'));
         $consumer->setPrefetchSize((int) $this->option('prefetch-size'));
         $consumer->setPrefetchCount((int) $this->option('prefetch-count'));
 

src/Console/QueueDeclareCommand.php

@@ -11,6 +11,7 @@ class QueueDeclareCommand extends Command
     protected $signature = 'rabbitmq:queue-declare
                            {name : The name of the queue to declare}
                            {connection=rabbitmq : The name of the queue connection to use}
+                           {--max-priority}
                            {--durable=1}
                            {--auto-delete=0}
                            {--quorum=0}';
@@ -33,9 +34,16 @@ public function handle(RabbitMQConnector $connector): void
             return;
         }
 
-        $arguments = (bool) $this->option('quorum')
-            ? ['x-queue-type' => 'quorum']
-            : [];
+        $arguments = [];
+
+        $maxPriority = (int) $this->option('max-priority');
+        if ($maxPriority) {
+            $arguments['x-max-priority'] = $maxPriority;
+        }
+
+        if($this->option('quorum')) {
+            $arguments['x-queue-type'] = 'quorum';
+        }
 
         $queue->declareQueue(
             $this->argument('name'),

src/Consumer.php

@@ -23,6 +23,9 @@ class Consumer extends Worker
     /** @var int */
     protected $prefetchSize;
 
+    /** @var int */
+    protected $maxPriority;
+
     /** @var int */
     protected $prefetchCount;
 
@@ -42,6 +45,11 @@ public function setConsumerTag(string $value): void
         $this->consumerTag = $value;
     }
 
+    public function setMaxPriority(int $value): void
+    {
+        $this->maxPriority = $value;
+    }
+
     public function setPrefetchSize(int $value): void
     {
         $this->prefetchSize = $value;
@@ -83,6 +91,10 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
         );
 
         $jobClass = $connection->getJobClass();
+        $arguments = [];
+        if ($this->maxPriority) {
+            $arguments['priority'] = ['I', $this->maxPriority];
+        }
 
         $this->channel->basic_consume(
             $queue,
@@ -113,7 +125,9 @@ function (AMQPMessage $message) use ($connection, $options, $connectionName, $qu
                 if ($this->supportsAsyncSignals()) {
                     $this->resetTimeoutHandler();
                 }
-            }
+            },
+            null,
+            $arguments
         );
 
         while ($this->channel->is_consuming()) {

src/Queue/RabbitMQQueue.php

@@ -532,14 +532,18 @@ protected function createMessage($payload, int $attempts = 0): array
             'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
         ];
 
-        if ($correlationId = json_decode($payload, true, 512)['id'] ?? null) {
+        $currentPayload = json_decode($payload, true, 512);
+        if ($correlationId = $currentPayload['id'] ?? null) {
             $properties['correlation_id'] = $correlationId;
         }
 
         if ($this->isPrioritizeDelayed()) {
             $properties['priority'] = $attempts;
         }
-
+        $commandData = unserialize($currentPayload['data']['command']);
+        if (property_exists($commandData, 'priority')) {
+            $properties['priority'] = $commandData->priority;
+        }
         $message = new AMQPMessage($payload, $properties);
 
         $message->set('application_headers', new AMQPTable([