Skip to content

[Doctrine Messenger] Fix support for pgsql + pgbouncer. #53819

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.

Sign up for GitHub

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

Merged
merged 1 commit into from
Feb 15, 2024
+351 −13
Merged

[Doctrine Messenger] Fix support for pgsql + pgbouncer. #53819

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions .github/workflows/integration-tests.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,14 @@ jobs:
sudo service redis-server restart
echo "::endgroup::"

- name: Install pgbouncer
run: |
sudo apt-get install -y pgbouncer
sudo cp src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Fixtures/pgbouncer/pgbouncer.ini /etc/pgbouncer/pgbouncer.ini
sudo cp src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Fixtures/pgbouncer/userlist.txt /etc/pgbouncer/userlist.txt
sudo service pgbouncer restart
sudo su - postgres -c "PGPASSWORD=password psql -Atq -h localhost -p 5432 -U postgres -d postgres -c \"SELECT usename, passwd FROM pg_shadow\""

- name: Configure Couchbase
run: |
curl -s -u 'username=Administrator&password=111111' -X POST http://localhost:8091/node/controller/setupServices -d 'services=kv%2Cn1ql%2Cindex%2Cfts'
Expand Down Expand Up @@ -186,6 +194,7 @@ jobs:
MESSENGER_SQS_FIFO_QUEUE_DSN: "sqs://localhost:4566/messages.fifo?sslmode=disable&poll_timeout=0.01"
KAFKA_BROKER: 127.0.0.1:9092
POSTGRES_HOST: localhost
PGBOUNCER_HOST: localhost:6432

#- name: Run HTTP push tests
# if: matrix.php == '8.1'
Expand Down
13 changes: 13 additions & 0 deletions src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Fixtures/pgbouncer/pgbouncer.ini
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
[databases]
postgres = host=localhost port=5432 user=postgres dbname=postgres pool_mode=transaction

[pgbouncer]
logfile = /var/log/postgresql/pgbouncer.log
pidfile = /var/run/postgresql/pgbouncer.pid
listen_addr = localhost
listen_port = 6432
unix_socket_dir = /var/run/postgresql
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt
max_client_conn = 20
default_pool_size = 20
1 change: 1 addition & 0 deletions src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Fixtures/pgbouncer/userlist.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"postgres" "md532e12f215ba27cb750c9e093ce4b5127"
98 changes: 98 additions & 0 deletions src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/ConnectionTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,104 @@ public function testItThrowsATransportExceptionIfItCannotRejectMessage()
$connection->reject('dummy_id');
}

public function testSend()
{
$queryBuilder = $this->getQueryBuilderMock();
$driverConnection = $this->getDBALConnectionMock();

$driverConnection->expects($this->once())
->method('createQueryBuilder')
->willReturn($queryBuilder);

$queryBuilder->expects($this->once())
->method('insert')
->willReturn($queryBuilder);

$queryBuilder->expects($this->once())
->method('values')
->with([
'body' => '?',
'headers' => '?',
'queue_name' => '?',
'created_at' => '?',
'available_at' => '?',
])
->willReturn($queryBuilder);

$queryBuilder->expects($this->once())
->method('getSQL')
->willReturn('INSERT');

$driverConnection->expects($this->once())
->method('beginTransaction');

$driverConnection->expects($this->once())
->method('executeStatement')
->with('INSERT')
->willReturn(1);

$driverConnection->expects($this->once())
->method('lastInsertId')
->willReturn('1');

$driverConnection->expects($this->once())
->method('commit');

$connection = new Connection([], $driverConnection);
$id = $connection->send('test', []);

self::assertSame('1', $id);
}

public function testSendLastInsertIdReturnsInteger()
{
$queryBuilder = $this->getQueryBuilderMock();
$driverConnection = $this->getDBALConnectionMock();

$driverConnection->expects($this->once())
->method('createQueryBuilder')
->willReturn($queryBuilder);

$queryBuilder->expects($this->once())
->method('insert')
->willReturn($queryBuilder);

$queryBuilder->expects($this->once())
->method('values')
->with([
'body' => '?',
'headers' => '?',
'queue_name' => '?',
'created_at' => '?',
'available_at' => '?',
])
->willReturn($queryBuilder);

$queryBuilder->expects($this->once())
->method('getSQL')
->willReturn('INSERT');

$driverConnection->expects($this->once())
->method('beginTransaction');

$driverConnection->expects($this->once())
->method('executeStatement')
->with('INSERT')
->willReturn(1);

$driverConnection->expects($this->once())
->method('lastInsertId')
->willReturn(1);

$driverConnection->expects($this->once())
->method('commit');

$connection = new Connection([], $driverConnection);
$id = $connection->send('test', []);

self::assertSame('1', $id);
}

private function getDBALConnectionMock()
{
$driverConnection = $this->createMock(DBALConnection::class);
Expand Down
89 changes: 89 additions & 0 deletions .../Messenger/Bridge/Doctrine/Tests/Transport/DoctrinePostgreSqlPgbouncerIntegrationTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
<?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\Configuration;
use Doctrine\DBAL\Connection;
use Doctrine\DBAL\DriverManager;
use Doctrine\DBAL\Schema\AbstractSchemaManager;
use Doctrine\DBAL\Schema\DefaultSchemaManagerFactory;
use Doctrine\DBAL\Tools\DsnParser;
use PHPUnit\Framework\TestCase;
use Symfony\Component\Messenger\Bridge\Doctrine\Tests\Fixtures\DummyMessage;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\PostgreSqlConnection;

/**
* This tests using PostgreSqlConnection with PgBouncer between pgsql and the application.
*
* @requires extension pdo_pgsql
*
* @group integration
*/
class DoctrinePostgreSqlPgbouncerIntegrationTest extends TestCase
{
private Connection $driverConnection;
private PostgreSqlConnection $connection;

public function testSendAndGetWithAutoSetupEnabledAndNotSetupAlready()
{
$this->connection->send('{"message": "Hi"}', ['type' => DummyMessage::class]);

$encoded = $this->connection->get();
$this->assertSame('{"message": "Hi"}', $encoded['body']);
$this->assertSame(['type' => DummyMessage::class], $encoded['headers']);

$this->assertNull($this->connection->get());
}

public function testSendAndGetWithAutoSetupEnabledAndSetupAlready()
{
$this->connection->setup();

$this->connection->send('{"message": "Hi"}', ['type' => DummyMessage::class]);

$encoded = $this->connection->get();
$this->assertSame('{"message": "Hi"}', $encoded['body']);
$this->assertSame(['type' => DummyMessage::class], $encoded['headers']);

$this->assertNull($this->connection->get());
}

protected function setUp(): void
{
if (!$host = getenv('PGBOUNCER_HOST')) {
$this->markTestSkipped('Missing PGBOUNCER_HOST env variable');
}

$url = "pdo-pgsql://postgres:password@$host";
$params = class_exists(DsnParser::class) ? (new DsnParser())->parse($url) : ['url' => $url];
$config = new Configuration();
if (class_exists(DefaultSchemaManagerFactory::class)) {
$config->setSchemaManagerFactory(new DefaultSchemaManagerFactory());
}

$this->driverConnection = DriverManager::getConnection($params, $config);
$this->connection = new PostgreSqlConnection(['table_name' => 'queue_table'], $this->driverConnection);
}

protected function tearDown(): void
{
$this->createSchemaManager()->dropTable('queue_table');
$this->driverConnection->close();
}

private function createSchemaManager(): AbstractSchemaManager
{
return method_exists($this->driverConnection, 'createSchemaManager')
? $this->driverConnection->createSchemaManager()
: $this->driverConnection->getSchemaManager();
}
}
89 changes: 89 additions & 0 deletions ...nt/Messenger/Bridge/Doctrine/Tests/Transport/DoctrinePostgreSqlRegularIntegrationTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
<?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\Configuration;
use Doctrine\DBAL\DriverManager;
use Doctrine\DBAL\Schema\AbstractSchemaManager;
use Doctrine\DBAL\Schema\DefaultSchemaManagerFactory;
use Doctrine\DBAL\Tools\DsnParser;
use PHPUnit\Framework\TestCase;
use Symfony\Component\Messenger\Bridge\Doctrine\Tests\Fixtures\DummyMessage;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\Connection;

/**
* This tests a using Doctrine PostgreSql connection without using PostgreSqlConnection
* that gets used when use_notify is enabled.
*
* @requires extension pdo_pgsql
*
* @group integration
*/
class DoctrinePostgreSqlRegularIntegrationTest extends TestCase
{
private \Doctrine\DBAL\Connection $driverConnection;
private Connection $connection;

public function testSendAndGetWithAutoSetupEnabledAndNotSetupAlready()
{
$this->connection->send('{"message": "Hi"}', ['type' => DummyMessage::class]);

$encoded = $this->connection->get();
$this->assertSame('{"message": "Hi"}', $encoded['body']);
$this->assertSame(['type' => DummyMessage::class], $encoded['headers']);

$this->assertNull($this->connection->get());
}

public function testSendAndGetWithAutoSetupEnabledAndSetupAlready()
{
$this->connection->setup();

$this->connection->send('{"message": "Hi"}', ['type' => DummyMessage::class]);

$encoded = $this->connection->get();
$this->assertSame('{"message": "Hi"}', $encoded['body']);
$this->assertSame(['type' => DummyMessage::class], $encoded['headers']);

$this->assertNull($this->connection->get());
}

protected function setUp(): void
{
if (!$host = getenv('POSTGRES_HOST')) {
$this->markTestSkipped('Missing POSTGRES_HOST env variable');
}

$url = "pdo-pgsql://postgres:password@$host";
$params = class_exists(DsnParser::class) ? (new DsnParser())->parse($url) : ['url' => $url];
$config = new Configuration();
if (class_exists(DefaultSchemaManagerFactory::class)) {
$config->setSchemaManagerFactory(new DefaultSchemaManagerFactory());
}

$this->driverConnection = DriverManager::getConnection($params, $config);
$this->connection = new Connection(['table_name' => 'queue_table'], $this->driverConnection);
}

protected function tearDown(): void
{
$this->createSchemaManager()->dropTable('queue_table');
$this->driverConnection->close();
}

private function createSchemaManager(): AbstractSchemaManager
{
return method_exists($this->driverConnection, 'createSchemaManager')
? $this->driverConnection->createSchemaManager()
: $this->driverConnection->getSchemaManager();
}
}
Loading