Always attempt to listen for notifications

symfony · fabpot · Aug 9, 2022 · Aug 6, 2022 · Aug 9, 2022 · 72ac5bfbe5d1b2e3be270dfb161c112456a32bed
commit 72ac5bfbe5d1b2e3be270dfb161c112456a32bed
diff --git a/src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/PostgreSqlConnectionTest.php b/src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/PostgreSqlConnectionTest.php
@@ -11,6 +11,11 @@

 namespace Symfony\Component\Messenger\Bridge\Doctrine\Tests\Transport;

+use Doctrine\DBAL\Cache\ArrayResult;
+use Doctrine\DBAL\Cache\ArrayStatement;
+use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
+use Doctrine\DBAL\Query\QueryBuilder;
+use Doctrine\DBAL\Result;
 use Doctrine\DBAL\Schema\Table;
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\Messenger\Bridge\Doctrine\Transport\PostgreSqlConnection;
@@ -42,6 +47,68 @@ public function testUnserialize()
        $connection->__wakeup();
    }

+    public function testListenOnConnection()
+    {
+        $driverConnection = $this->createMock(\Doctrine\DBAL\Connection::class);
+
+        $driverConnection
+            ->expects(self::any())
+            ->method('getDatabasePlatform')
+            ->willReturn(new PostgreSQLPlatform());
+
+        $driverConnection
+            ->expects(self::any())
+            ->method('createQueryBuilder')
+            ->willReturn(new QueryBuilder($driverConnection));
+
+        $wrappedConnection = new class() {
+            private $notifyCalls = 0;
+
+            public function pgsqlGetNotify()
+            {
+                ++$this->notifyCalls;
+
+                return false;
+            }
+
+            public function countNotifyCalls()
+            {
+                return $this->notifyCalls;
+            }
+        };
+
+        // dbal 2.x
+        if (interface_exists(Result::class)) {
+            $driverConnection
+                ->expects(self::exactly(2))
+                ->method('getWrappedConnection')
+                ->willReturn($wrappedConnection);
+
+            $driverConnection
+                ->expects(self::any())
+                ->method('executeQuery')
+                ->willReturn(new ArrayStatement([]));
+        } else {
+            // dbal 3.x
+            $driverConnection
+                ->expects(self::exactly(2))
+                ->method('getNativeConnection')
+                ->willReturn($wrappedConnection);
+
+            $driverConnection
+                ->expects(self::any())
+                ->method('executeQuery')
+                ->willReturn(new Result(new ArrayResult([]), $driverConnection));
+        }
+        $connection = new PostgreSqlConnection(['table_name' => 'queue_table'], $driverConnection);
+
+        $connection->get(); // first time we have queueEmptiedAt === null, fallback on the parent implementation
+        $connection->get();
+        $connection->get();
+
+        $this->assertSame(2, $wrappedConnection->countNotifyCalls());
+    }
+
    public function testGetExtraSetupSql()
    {
        $driverConnection = $this->createMock(\Doctrine\DBAL\Connection::class);

diff --git a/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/PostgreSqlConnection.php b/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/PostgreSqlConnection.php
@@ -33,8 +33,6 @@ final class PostgreSqlConnection extends Connection
        'get_notify_timeout' => 0,
    ];

-    private $listening = false;
-
    public function __sleep(): array
    {
        throw new \BadMethodCallException('Cannot serialize '.__CLASS__);
@@ -62,12 +60,9 @@ public function get(): ?array
            return parent::get();
        }

-        if (!$this->listening) {
-            // This is secure because the table name must be a valid identifier:
-            // https://www.postgresql.org/docs/current/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS
-            $this->executeStatement(sprintf('LISTEN "%s"', $this->configuration['table_name']));
-            $this->listening = true;
-        }
+        // This is secure because the table name must be a valid identifier:
+        // https://www.postgresql.org/docs/current/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS
+        $this->executeStatement(sprintf('LISTEN "%s"', $this->configuration['table_name']));

        if (method_exists($this->driverConnection, 'getNativeConnection')) {
            $wrappedConnection = $this->driverConnection->getNativeConnection();
@@ -150,11 +145,6 @@ private function createTriggerFunctionName(): string

    private function unlisten()
    {
-        if (!$this->listening) {
-            return;
-        }
-
        $this->executeStatement(sprintf('UNLISTEN "%s"', $this->configuration['table_name']));
-        $this->listening = false;
    }
 }