Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Messenger] Add support for PostgreSQL LISTEN/NOTIFY
- Loading branch information
Showing
7 changed files
with
206 additions
and
19 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,3 +5,4 @@ CHANGELOG | |
| ----- | ||
|
|
||
| * Introduced the Doctrine bridge. | ||
| * Added support for PostgreSQL `LISTEN`/`NOTIFY`. | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
46 changes: 46 additions & 0 deletions
46
src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/PostgreSqlConnectionTest.php
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| <?php | ||
|
|
||
| /* | ||
| * This file is part of the Symfony package. | ||
| * | ||
| * (c) Fabien Potencier <fabien@symfony.com> | ||
| * | ||
| * For the full copyright and license information, please view the LICENSE | ||
| * file that was distributed with this source code. | ||
| */ | ||
|
|
||
| namespace Symfony\Component\Messenger\Bridge\Doctrine\Tests\Transport; | ||
|
|
||
| use Doctrine\DBAL\Schema\Synchronizer\SchemaSynchronizer; | ||
| use PHPUnit\Framework\TestCase; | ||
| use Symfony\Component\Messenger\Bridge\Doctrine\Transport\PostgreSqlConnection; | ||
|
|
||
| /** | ||
| * @author Kévin Dunglas <dunglas@gmail.com> | ||
| */ | ||
| class PostgreSqlConnectionTest extends TestCase | ||
| { | ||
| public function testSerialize() | ||
| { | ||
| $this->expectException(\BadMethodCallException::class); | ||
| $this->expectExceptionMessage('Cannot serialize '.PostgreSqlConnection::class); | ||
|
|
||
| $schemaSynchronizer = $this->createMock(SchemaSynchronizer::class); | ||
| $driverConnection = $this->createMock(\Doctrine\DBAL\Connection::class); | ||
|
|
||
| $connection = new PostgreSqlConnection([], $driverConnection, $schemaSynchronizer); | ||
| serialize($connection); | ||
| } | ||
|
|
||
| public function testUnserialize() | ||
| { | ||
| $this->expectException(\BadMethodCallException::class); | ||
| $this->expectExceptionMessage('Cannot unserialize '.PostgreSqlConnection::class); | ||
|
|
||
| $schemaSynchronizer = $this->createMock(SchemaSynchronizer::class); | ||
| $driverConnection = $this->createMock(\Doctrine\DBAL\Connection::class); | ||
|
|
||
| $connection = new PostgreSqlConnection([], $driverConnection, $schemaSynchronizer); | ||
| $connection->__wakeup(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
120 changes: 120 additions & 0 deletions
120
src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/PostgreSqlConnection.php
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,120 @@ | ||
| <?php | ||
|
|
||
| /* | ||
| * This file is part of the Symfony package. | ||
| * | ||
| * (c) Fabien Potencier <fabien@symfony.com> | ||
| * | ||
| * For the full copyright and license information, please view the LICENSE | ||
| * file that was distributed with this source code. | ||
| */ | ||
|
|
||
| declare(strict_types=1); | ||
|
|
||
| namespace Symfony\Component\Messenger\Bridge\Doctrine\Transport; | ||
|
|
||
| /** | ||
| * Uses PostgreSQL LISTEN/NOTIFY to push messages to workers. | ||
| * | ||
| * @internal | ||
| * @final | ||
| * | ||
| * @author Kévin Dunglas <dunglas@gmail.com> | ||
| */ | ||
| class PostgreSqlConnection extends Connection | ||
| { | ||
| /** | ||
| * * use_notify: Set to false to disable the use of LISTEN/NOTIFY. Default: true | ||
| * * check_delayed_interval: The interval to check for delayed messages, in milliseconds. Set to 0 to disable checks. Default: 1000 | ||
| * * get_notify_timeout: The length of time to wait for a response when calling PDO::pgsqlGetNotify, in milliseconds. Default: 0. | ||
| */ | ||
| protected const DEFAULT_OPTIONS = parent::DEFAULT_OPTIONS + [ | ||
| 'check_delayed_interval' => 1000, | ||
| 'get_notify_timeout' => 0, | ||
| ]; | ||
|
|
||
| private $listening = false; | ||
|
|
||
| public function __sleep() | ||
| { | ||
| throw new \BadMethodCallException('Cannot serialize '.__CLASS__); | ||
| } | ||
|
|
||
| public function __wakeup() | ||
| { | ||
| throw new \BadMethodCallException('Cannot unserialize '.__CLASS__); | ||
| } | ||
|
|
||
| public function __destruct() | ||
| { | ||
| $this->unlisten(); | ||
| } | ||
|
|
||
| public function reset() | ||
| { | ||
| parent::reset(); | ||
| $this->unlisten(); | ||
| } | ||
|
|
||
| public function get(): ?array | ||
| { | ||
| if (null === $this->queueEmptiedAt) { | ||
| 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->driverConnection->exec(sprintf('LISTEN "%s"', $this->configuration['table_name'])); | ||
| $this->listening = true; | ||
| } | ||
|
|
||
| $notification = $this->driverConnection->getWrappedConnection()->pgsqlGetNotify(\PDO::FETCH_ASSOC, $this->configuration['get_notify_timeout']); | ||
| if ( | ||
| // no notifications, or for another table or queue | ||
| (false === $notification || $notification['message'] !== $this->configuration['table_name'] || $notification['payload'] !== $this->configuration['queue_name']) && | ||
| // delayed messages | ||
| (microtime(true) * 1000 - $this->queueEmptiedAt < $this->configuration['check_delayed_interval']) | ||
| ) { | ||
| return null; | ||
| } | ||
|
|
||
| return parent::get(); | ||
| } | ||
|
|
||
| public function setup(): void | ||
| { | ||
| parent::setup(); | ||
|
|
||
| $sql = sprintf(<<<'SQL' | ||
| LOCK TABLE %1$s; | ||
| -- create trigger function | ||
| CREATE OR REPLACE FUNCTION notify_%1$s() RETURNS TRIGGER AS $$ | ||
| BEGIN | ||
| PERFORM pg_notify('%1$s', NEW.queue_name::text); | ||
| RETURN NEW; | ||
| END; | ||
| $$ LANGUAGE plpgsql; | ||
| -- register trigger | ||
| DROP TRIGGER IF EXISTS notify_trigger ON %1$s; | ||
| CREATE TRIGGER notify_trigger | ||
| AFTER INSERT | ||
| ON %1$s | ||
| FOR EACH ROW EXECUTE PROCEDURE notify_%1$s(); | ||
| SQL | ||
| , $this->configuration['table_name']); | ||
| $this->driverConnection->exec($sql); | ||
| } | ||
|
|
||
| private function unlisten() | ||
| { | ||
| if (!$this->listening) { | ||
| return; | ||
| } | ||
|
|
||
| $this->driverConnection->exec(sprintf('UNLISTEN "%s"', $this->configuration['table_name'])); | ||
| $this->listening = false; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters