There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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 = $th 8000 is->createMock(SchemaSynchronizer::class); | ||
| $driverConnection = $this->createMock(\Doctrine\DBAL\Connection::class); | ||
|
|
||
| $connection = new PostgreSqlConnection([], $driverConnection, $schemaSynchronizer); | ||
| $connection->__wakeup(); | ||
| } | ||
| } |
This file contains hidden or 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 hidden or 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 hidden or 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 | ||
dunglas marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| { | ||
| /** | ||
| * * 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 hidden or 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
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
[Messenger] Add support for PostgreSQL LISTEN/NOTIFY #35485
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[Messenger] Add support for PostgreSQL LISTEN/NOTIFY #35485
Changes from all commits
01f33c3File filter
Filter by extension
Conversations
Jump to