Fixed losing message priority for dbal driver

Author: vtsykun

pkg/dbal/Client/DbalDriver.php

@@ -30,6 +30,17 @@ class DbalDriver implements DriverInterface
      */
     private $queueMetaRegistry;
 
+    /**
+     * @var array
+     */
+    private static $priorityMap = [
+        MessagePriority::VERY_LOW => 0,
+        MessagePriority::LOW => 1,
+        MessagePriority::NORMAL => 2,
+        MessagePriority::HIGH => 3,
+        MessagePriority::VERY_HIGH => 4,
+    ];
+
     /**
      * @param DbalContext       $context
      * @param Config            $config
@@ -63,6 +74,9 @@ public function createTransportMessage(Message $message)
         $transportMessage->setDelay($message->getDelay());
         $transportMessage->setReplyTo($message->getReplyTo());
         $transportMessage->setCorrelationId($message->getCorrelationId());
+        if (array_key_exists($message->getPriority(), self::$priorityMap)) {
+            $transportMessage->setPriority(self::$priorityMap[$message->getPriority()]);
+        }
 
         return $transportMessage;
     }
@@ -83,11 +97,17 @@ public function createClientMessage(PsrMessage $message)
         $clientMessage->setContentType($message->getHeader('content_type'));
         $clientMessage->setMessageId($message->getMessageId());
         $clientMessage->setTimestamp($message->getTimestamp());
-        $clientMessage->setPriority(MessagePriority::NORMAL);
         $clientMessage->setDelay($message->getDelay());
         $clientMessage->setReplyTo($message->getReplyTo());
         $clientMessage->setCorrelationId($message->getCorrelationId());
 
+        $priorityMap = array_flip(self::$priorityMap);
+        $priority = array_key_exists($message->getPriority(), $priorityMap) ?
+            $priorityMap[$message->getPriority()] :
+            MessagePriority::NORMAL;
+        $clientMessage->setPriority($priority);
+
+
         return $clientMessage;
     }
 
@@ -156,4 +176,12 @@ public function getConfig()
     {
         return $this->config;
     }
+
+    /**
+     * @return array
+     */
+    public static function getPriorityMap()
+    {
+        return self::$priorityMap;
+    }
 }

pkg/dbal/DbalConsumer.php

@@ -176,7 +176,7 @@ protected function receiveMessage()
                 ->where('queue = :queue')
                 ->andWhere('(delayed_until IS NULL OR delayed_until <= :delayedUntil)')
                 ->orderBy('priority', 'desc')
-                ->orderBy('id', 'asc')
+                ->addOrderBy('id', 'asc')
                 ->setMaxResults(1)
             ;
 

pkg/dbal/Tests/DbalConsumerTest.php

@@ -172,10 +172,15 @@ public function testShouldReceiveMessage()
             ->will($this->returnSelf())
         ;
         $queryBuilder
-            ->expects($this->exactly(2))
+            ->expects($this->exactly(1))
             ->method('orderBy')
             ->will($this->returnSelf())
         ;
+        $queryBuilder
+            ->expects($this->exactly(1))
+            ->method('addOrderBy')
+            ->will($this->returnSelf())
+        ;
 
         $platform = $this->createPlatformMock();
 
@@ -264,10 +269,15 @@ public function testShouldReturnNullIfThereIsNoNewMessage()
             ->will($this->returnSelf())
         ;
         $queryBuilder
-            ->expects($this->exactly(2))
+            ->expects($this->exactly(1))
             ->method('orderBy')
             ->will($this->returnSelf())
         ;
+        $queryBuilder
+            ->expects($this->exactly(1))
+            ->method('addOrderBy')
+            ->will($this->returnSelf())
+        ;
 
         $platform = $this->createPlatformMock();
 
@@ -352,10 +362,15 @@ public function testShouldThrowIfMessageWasNotRemoved()
             ->will($this->returnSelf())
         ;
         $queryBuilder
-            ->expects($this->exactly(2))
+            ->expects($this->exactly(1))
             ->method('orderBy')
             ->will($this->returnSelf())
         ;
+        $queryBuilder
+            ->expects($this->exactly(1))
+            ->method('addOrderBy')
+            ->will($this->returnSelf())
+        ;
 
         $platform = $this->createPlatformMock();