diogo
/
symfony
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

bug #33127 [Messenger] make delay exchange and queues durable like the normal ones by default (Tobion) 

This PR was merged into the 4.3 branch.

Discussion
----------

[Messenger] make delay exchange and queues durable like the normal ones by default

| Q             | A
| ------------- | ---
| Branch?       | 4.3
| Bug fix?      | yes
| New feature?  | no <!-- please update src/**/CHANGELOG.md files -->
| BC breaks?    | no     <!-- see https://symfony.com/bc -->
| Deprecations? | no <!-- please update UPGRADE-*.md and src/**/CHANGELOG.md files -->
| Tests pass?   | yes    <!-- please add some, will be required by reviewers -->
| Fixed tickets | #32891
| License       | MIT
| Doc PR        |

This also imrproves BC of #32631 by only adding the new expiry argument in case the delay queue name was not overwritten using the options. I will remove the checks in 4.4 again. Please merge this PR before releasing the new 4.3 version so that 32631 and this PR are part of one release.

Commits
-------

e5ecda6de1 [Messenger] make delay exchange and queues durable like the normal ones by default
This commit is contained in:
Fabien Potencier 2019-08-18 10:19:16 +02:00
parent a553173439 e5ecda6de1
commit 431ead273a
2 changed files with 23 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/Symfony/Component/Messenger/Tests/Transport/AmqpExt/ConnectionTest.php
View File

@ -337,7 +337,7 @@ class ConnectionTest extends TestCase
$amqpQueue->expects($this->once())->method('setName')->with(self::DEFAULT_EXCHANGE_NAME);
$amqpQueue->expects($this->once())->method('declareQueue');
$delayExchange->expects($this->once())->method('setName')->with('delay');
$delayExchange->expects($this->once())->method('setName')->with('delays');
$delayExchange->expects($this->once())->method('declareExchange');
$delayExchange->expects($this->once())->method('publish');
@ -371,7 +371,7 @@ class ConnectionTest extends TestCase
]);
$delayQueue->expects($this->once())->method('declareQueue');
$delayQueue->expects($this->once())->method('bind')->with('delay', 'delay_messages__5000');
$delayQueue->expects($this->once())->method('bind')->with('delays', 'delay_messages__5000');
$delayExchange->expects($this->once())->method('publish')->with('{}', 'delay_messages__5000', AMQP_NOPARAM, ['headers' => ['x-some-headers' => 'foo']]);
@ -413,7 +413,7 @@ class ConnectionTest extends TestCase
]);
$delayQueue->expects($this->once())->method('declareQueue');
$delayQueue->expects($this->once())->method('bind')->with('delay', 'delay_messages__120000');
$delayQueue->expects($this->once())->method('bind')->with('delays', 'delay_messages__120000');
$delayExchange->expects($this->once())->method('publish')->with('{}', 'delay_messages__120000', AMQP_NOPARAM, ['headers' => []]);
$connection->publish('{}', [], 120000);
@ -517,7 +517,7 @@ class ConnectionTest extends TestCase
]);
$delayQueue->expects($this->once())->method('declareQueue');
$delayQueue->expects($this->once())->method('bind')->with('delay', 'delay_messages_routing_key_120000');
$delayQueue->expects($this->once())->method('bind')->with('delays', 'delay_messages_routing_key_120000');
$delayExchange->expects($this->once())->method('publish')->with('{}', 'delay_messages_routing_key_120000', AMQP_NOPARAM, ['headers' => []]);
$connection->publish('{}', [], 120000, new AmqpStamp('routing_key'));

25
src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php
View File

@ -62,7 +62,7 @@ class Connection
{
$this->connectionOptions = array_replace_recursive([
'delay' => [
'exchange_name' => 'delay',
'exchange_name' => 'delays',
'queue_name_pattern' => 'delay_%exchange_name%_%routing_key%_%delay%',
],
], $connectionOptions);
@ -93,7 +93,7 @@ class Connection
* * arguments: Extra arguments
* * delay:
* * queue_name_pattern: Pattern to use to create the queues (Default: "delay_%exchange_name%_%routing_key%_%delay%")
* * exchange_name: Name of the exchange to be used for the delayed/retried messages (Default: "delay")
* * exchange_name: Name of the exchange to be used for the delayed/retried messages (Default: "delays")
* * auto_setup: Enable or not the auto-setup of queues and exchanges (Default: true)
* * prefetch_count: set channel prefetch count
*/
@ -252,6 +252,11 @@ class Connection
$this->amqpDelayExchange = $this->amqpFactory->createExchange($this->channel());
$this->amqpDelayExchange->setName($this->connectionOptions['delay']['exchange_name']);
$this->amqpDelayExchange->setType(AMQP_EX_TYPE_DIRECT);
if ('delays' === $this->connectionOptions['delay']['exchange_name']) {
// only add the new flag when the name was not provided explicitly so we're using the new default name to prevent a redeclaration error
// the condition will be removed in 4.4
$this->amqpDelayExchange->setFlags(AMQP_DURABLE);
}
}
return $this->amqpDelayExchange;
@ -274,16 +279,24 @@ class Connection
[$delay, $this->exchangeOptions['name'], $routingKey ?? ''],
$this->connectionOptions['delay']['queue_name_pattern']
));
if ('delay_%exchange_name%_%routing_key%_%delay%' === $this->connectionOptions['delay']['queue_name_pattern']) {
// the condition will be removed in 4.4
$queue->setFlags(AMQP_DURABLE);
$extraArguments = [
// delete the delay queue 10 seconds after the message expires
// publishing another message redeclares the queue which renews the lease
'x-expires' => $delay + 10000,
];
} else {
$extraArguments = [];
}
$queue->setArguments([
'x-message-ttl' => $delay,
// delete the delay queue 10 seconds after the message expires
// publishing another message redeclares the queue which renews the lease
'x-expires' => $delay + 10000,
'x-dead-letter-exchange' => $this->exchangeOptions['name'],
// after being released from to DLX, make sure the original routing key will be used
// we must use an empty string instead of null for the argument to be picked up
'x-dead-letter-routing-key' => $routingKey ?? '',
]);
] + $extraArguments);
return $queue;
}