Added write lock

Author: rosamarsky

Diff for: pkg/dbal/DbalConsumerHelperTrait.php

@@ -20,20 +20,22 @@ protected function fetchMessage(array $queues, string $deliveryId, int $redelive
 
             $this->getConnection()->beginTransaction();
 
-            $message = $this->getConnection()->createQueryBuilder()
+            $query = $this->getConnection()->createQueryBuilder()
                 ->select('*')
                 ->from($this->getContext()->getTableName())
                 ->andWhere('delivery_id IS NULL')
                 ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil')
                 ->andWhere('queue IN (:queues)')
                 ->addOrderBy('priority', 'desc')
                 ->addOrderBy('published_at', 'asc')
-                ->setParameter('delayedUntil', $now, \Doctrine\DBAL\ParameterType::INTEGER)
-                ->setParameter('queues', array_values($queues), \Doctrine\DBAL\Connection::PARAM_STR_ARRAY)
-                ->setMaxResults(1)
-                ->execute()
-                ->fetch()
-            ;
+                ->setMaxResults(1);
+
+            // select for update
+            $message = $this->getConnection()->executeQuery(
+                $query->getSQL().' '.$this->getConnection()->getDatabasePlatform()->getWriteLockSQL(),
+                ['delayedUntil' => $now, 'queues' => array_values($queues)],
+                ['delayedUntil' => \Doctrine\DBAL\ParameterType::INTEGER, 'queues' => \Doctrine\DBAL\Connection::PARAM_STR_ARRAY]
+            )->fetch();
 
             if (!$message) {
                 $this->getConnection()->commit();
@@ -50,7 +52,8 @@ protected function fetchMessage(array $queues, string $deliveryId, int $redelive
                 ->setParameter('id', $message['id'], Type::GUID)
                 ->setParameter('deliveryId', $deliveryId, Type::STRING)
                 ->setParameter('redeliverAfter', $now + $redeliveryDelay, Type::BIGINT)
-                ->execute();
+                ->execute()
+            ;
 
             $this->getConnection()->commit();
 
@@ -61,7 +64,8 @@ protected function fetchMessage(array $queues, string $deliveryId, int $redelive
                 ->setParameter('deliveryId', $deliveryId, Type::STRING)
                 ->setMaxResults(1)
                 ->execute()
-                ->fetch();
+                ->fetch()
+            ;
 
             return $deliveredMessage ?: null;
         } catch (\Exception $e) {

Diff for: pkg/dbal/Tests/Functional/DbalConsumerTest.php

@@ -49,7 +49,6 @@ public function testShouldSetPublishedAtDateToReceivedMessage()
 
         $producer = $context->createProducer();
 
-        /** @var DbalMessage $message */
         $message = $context->createMessage($expectedBody);
         $message->setPublishedAt($time);
         $producer->send($queue, $message);
@@ -62,7 +61,7 @@ public function testShouldSetPublishedAtDateToReceivedMessage()
         $this->assertSame($time, $message->getPublishedAt());
     }
 
-    public function testShouldOrderMessagesWithSamePriorityByPublishedAtDateee()
+    public function testShouldOrderMessagesWithSamePriorityByPublishedAtDate()
     {
         $context = $this->context;
         $queue = $context->createQueue(__METHOD__);
@@ -80,7 +79,6 @@ public function testShouldOrderMessagesWithSamePriorityByPublishedAtDateee()
 
         $producer = $context->createProducer();
 
-        /** @var DbalMessage $message */
         $message = $context->createMessage($expectedPriority5Body);
         $message->setPriority(5);
         $message->setPublishedAt($time);