Skip to content

[Messenger] Ability to use transactions while sending messages with AMQP #51719

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

Open
wants to merge 7 commits into
base: 7.4
Choose a base branch
from
Open

[Messenger] Ability to use transactions while sending messages with AMQP #51719

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6064a73
[Messenger] Add option option `transactional` to publish on AMQP queu…
michalsitek Sep 22, 2023
14de41d
Fixed changed version in changelog
michalsitek Sep 22, 2023
f290f46
Removed redundant sprintf
michalsitek Sep 22, 2023
e58dfd0
Removed redundant newline
michalsitek Sep 22, 2023
6c81d71
Made sure transaction is started before and committed after publishing
michalsitek Sep 22, 2023
ecb8749
Added missing return value for mock in tests
michalsitek Sep 22, 2023
41a9f2c
Removed double newlines
michalsitek Sep 22, 2023
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
5 changes: 5 additions & 0 deletions src/Symfony/Component/Messenger/Bridge/Amqp/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
CHANGELOG
=========

6.4
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
6.4
7.3

---

* Add option to use transactions for publishing

6.0
---

Expand Down
63 changes: 63 additions & 0 deletions src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/ConnectionTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,12 @@ public function testExceptionIfInvalidExchangeOptionIsPassed()
Connection::fromDsn('amqp://host', ['exchange' => ['foo' => 'bar']]);
}

public function testExceptionIfConfirmTimeoutAndTransactionalIsPassed()
{
$this->expectExceptionMessage('Confirm timeout cannot be used on transactional channel.');
Connection::fromDsn('amqp://host?confirm_timeout=0.5&transactional=1');
}

public function testSetsParametersOnTheQueueAndExchange()
{
$factory = new TestAmqpFactory(
Expand Down Expand Up @@ -753,6 +759,63 @@ public function testItCanPublishAndWaitForConfirmation()
$connection->publish('body');
}

public function testItPublishMessagesWithoutTransactions()
{
$factory = new TestAmqpFactory(
$this->createMock(\AMQPConnection::class),
$amqpChannel = $this->createMock(\AMQPChannel::class),
$this->createMock(\AMQPQueue::class),
$amqpExchange = $this->createMock(\AMQPExchange::class)
);

$amqpExchange->method('getChannel')->willReturn($amqpChannel);

$amqpChannel->expects($this->never())->method('startTransaction');
$amqpChannel->expects($this->never())->method('commitTransaction');

$connection = Connection::fromDsn('amqp://localhost', [], $factory);
$connection->publish('body');
}

public function testItPublishMessagesUsingTransactions()
{
$factory = new TestAmqpFactory(
$this->createMock(\AMQPConnection::class),
$amqpChannel = $this->createMock(\AMQPChannel::class),
$this->createMock(\AMQPQueue::class),
$amqpExchange = $this->createMock(\AMQPExchange::class)
);

$amqpExchange->method('getChannel')->willReturn($amqpChannel);

$startedTransaction = false;
$amqpChannel->expects($this->once())->method('startTransaction')->willReturnCallback(
function () use (&$startedTransaction) {
$startedTransaction = true;
}
);

$published = false;
$amqpExchange->expects($this->once())->method('publish')->with(
'body',
null,
\AMQP_NOPARAM,
['headers' => [], 'delivery_mode' => 2, 'timestamp' => time()]
)->willReturnCallback(function () use (&$startedTransaction, &$published) {
$this->assertTrue($startedTransaction);
$published = true;
});

$amqpChannel->expects($this->once())->method('commitTransaction')->willReturnCallback(
function () use (&$published) {
$this->assertTrue($published);
}
);

$connection = Connection::fromDsn('amqp://localhost?transactional=1', [], $factory);
$connection->publish('body');
}

public function testItCanBeConstructedWithTLSOptionsAndNonTLSDsn()
{
$this->assertEquals(
Expand Down
18 changes: 18 additions & 0 deletions src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ class Connection
'read_timeout',
'write_timeout',
'confirm_timeout',
'transactional',
'connect_timeout',
'rpc_timeout',
'cacert',
Expand Down Expand Up @@ -128,6 +129,7 @@ public function __construct(array $connectionOptions, array $exchangeOptions, ar
* * write_timeout: Timeout in for outcome activity. Note: 0 or greater seconds. May be fractional.
* * connect_timeout: Connection timeout. Note: 0 or greater seconds. May be fractional.
* * confirm_timeout: Timeout in seconds for confirmation, if none specified transport will not wait for message confirmation. Note: 0 or greater seconds. May be fractional.
* * transactional: Enable or not using transactions for publishing (Default: false)
* * queues[name]: An array of queues, keyed by the name
* * binding_keys: The binding keys (if any) to bind to this queue
* * binding_arguments: Arguments to be used while binding the queue.
Expand Down Expand Up @@ -251,6 +253,13 @@ private static function validateOptions(array $options): void
&& 0 < \count($invalidExchangeOptions = array_diff(array_keys($options['exchange']), self::AVAILABLE_EXCHANGE_OPTIONS))) {
throw new LogicException(sprintf('Invalid exchange option(s) "%s" passed to the AMQP Messenger transport.', implode('", "', $invalidExchangeOptions)));
}

$transactional = isset($options['transactional']) && filter_var($options['transactional'], \FILTER_VALIDATE_BOOL);
$confirmTimeout = ('' !== ($options['confirm_timeout'] ?? ''));

if ($transactional && $confirmTimeout) {
throw new LogicException('Confirm timeout cannot be used on transactional channel.');
}
}

private static function normalizeQueueArguments(array $arguments): array
Expand Down Expand Up @@ -334,16 +343,25 @@ private function publishOnExchange(\AMQPExchange $exchange, string $body, string
$attributes['headers'] = array_merge($attributes['headers'] ?? [], $headers);
$attributes['delivery_mode'] ??= 2;
$attributes['timestamp'] ??= time();
$transactional = isset($this->connectionOptions['transactional']) && filter_var($this->connectionOptions['transactional'], \FILTER_VALIDATE_BOOL);

$this->lastActivityTime = time();

if ($transactional) {
$exchange->getChannel()->startTransaction();
}

$exchange->publish(
$body,
$routingKey,
$amqpStamp ? $amqpStamp->getFlags() : \AMQP_NOPARAM,
$attributes
);

if ($transactional) {
$exchange->getChannel()->commitTransaction();
}

if ('' !== ($this->connectionOptions['confirm_timeout'] ?? '')) {
$this->channel()->waitForConfirm((float) $this->connectionOptions['confirm_timeout']);
}
Expand Down