fix compatibility with Doctrine DBAL 4

Author: xabbuh

src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/ConnectionTest.php

@@ -82,6 +82,9 @@ public function testGetWithNoPendingMessageWillReturnNull()
         $queryBuilder
             ->method('getParameterTypes')
             ->willReturn([]);
+        $queryBuilder
+            ->method('getSQL')
+            ->willReturn('SELECT FOR UPDATE');
         $driverConnection->expects($this->once())
             ->method('createQueryBuilder')
             ->willReturn($queryBuilder);
@@ -120,7 +123,11 @@ private function getDBALConnectionMock()
     {
         $driverConnection = $this->createMock(DBALConnection::class);
         $platform = $this->createMock(AbstractPlatform::class);
-        $platform->method('getWriteLockSQL')->willReturn('FOR UPDATE');
+
+        if (!method_exists(QueryBuilder::class, 'forUpdate')) {
+            $platform->method('getWriteLockSQL')->willReturn('FOR UPDATE');
+        }
+
         $configuration = $this->createMock(\Doctrine\DBAL\Configuration::class);
         $driverConnection->method('getDatabasePlatform')->willReturn($platform);
         $driverConnection->method('getConfiguration')->willReturn($configuration);
@@ -381,7 +388,9 @@ public function testGeneratedSql(AbstractPlatform $platform, string $expectedSql
         $driverConnection
             ->expects($this->once())
             ->method('executeQuery')
-            ->with($expectedSql)
+            ->with($this->callback(function ($sql) use ($expectedSql) {
+                return trim($expectedSql) === trim($sql);
+            }))
             ->willReturn($result)
         ;
         $driverConnection->expects($this->once())->method('commit');
@@ -415,6 +424,12 @@ class_exists(SQLServerPlatform::class) && !class_exists(SQLServer2012Platform::c
                 new OraclePlatform(),
                 'SELECT w.id AS "id", w.body AS "body", w.headers AS "headers", w.queue_name AS "queue_name", w.created_at AS "created_at", w.available_at AS "available_at", w.delivered_at AS "delivered_at" FROM messenger_messages w WHERE w.id IN (SELECT m.id FROM messenger_messages m WHERE (m.queue_name = ?) AND (m.delivered_at is null OR m.delivered_at < ?) AND (m.available_at <= ?) ORDER BY available_at ASC FETCH NEXT 1 ROWS ONLY) FOR UPDATE',
             ];
+        } elseif (method_exists(QueryBuilder::class, 'forUpdate')) {
+            // DBAL >= 3.8
+            yield 'Oracle' => [
+                new OraclePlatform(),
+                'SELECT w.id AS "id", w.body AS "body", w.headers AS "headers", w.queue_name AS "queue_name", w.created_at AS "created_at", w.available_at AS "available_at", w.delivered_at AS "delivered_at" FROM messenger_messages w WHERE w.id IN (SELECT a.id FROM (SELECT m.id FROM messenger_messages m WHERE (m.queue_name = ?) AND (m.delivered_at is null OR m.delivered_at < ?) AND (m.available_at <= ?) ORDER BY available_at ASC) a WHERE ROWNUM <= 1 FOR UPDATE)',
+            ];
         } else {
             // DBAL < 4
             yield 'Oracle' => [

src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/Connection.php

@@ -177,7 +177,14 @@ public function get(): ?array
 
             // Append pessimistic write lock to FROM clause if db platform supports it
             $sql = $query->getSQL();
-            if (preg_match('/FROM (.+) WHERE/', (string) $sql, $matches)) {
+
+            if (method_exists(QueryBuilder::class, 'forUpdate')) {
+                $query->forUpdate();
+                try {
+                    $sql = $query->getSQL();
+                } catch (DBALException $e) {
+                }
+            } elseif (preg_match('/FROM (.+) WHERE/', (string) $sql, $matches)) {
                 $fromClause = $matches[1];
                 $sql = str_replace(
                     sprintf('FROM %s WHERE', $fromClause),
@@ -194,8 +201,12 @@ public function get(): ?array
             }
 
             // use SELECT ... FOR UPDATE to lock table
+            if (!method_exists(QueryBuilder::class, 'forUpdate')) {
+                $sql .= ' '.$this->driverConnection->getDatabasePlatform()->getWriteLockSQL();
+            }
+
             $stmt = $this->executeQuery(
-                $sql.' '.$this->driverConnection->getDatabasePlatform()->getWriteLockSQL(),
+                $sql,
                 $query->getParameters(),
                 $query->getParameterTypes()
             );