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

bug #32035 [Messenger] fix delay delivery for non-fanout exchanges (Tobion)

Browse Source 
This PR was merged into the 4.3 branch.

Discussion
----------

[Messenger] fix delay delivery for non-fanout exchanges

| 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 |
| License       | MIT
| Doc PR        | symfony/symfony-docs#... <!-- required for new features -->

Fixes two bugs and outdated phpdoc:
1. When your exchange is not of type fanout, but direct for example, then delivery of delayed (retrying) messages does not work. This is because the delay logics adds a routing key to message. It was fixed if you have a custom routing key in #31355. But if you have no routing key, it still changed the routing key which means the message will not be delivery from your direct exchange to your queue anymore after being in the delay exchange. For fanout, which is the default, it does not matter because the routing key is ignored.
2. also fix dsn parsing of plain `amqp://` which is a valid URI that parse_url cannot handle when you want to pass all parameters as options

Commits
-------

0f15306d61 [Messenger] fix delay delivery for non-fanout exchanges
This commit is contained in:
Tobias Schultze 2019-06-14 19:11:46 +02:00
commit a9bcdcc936
2 changed files with 54 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -25,14 +25,14 @@ class ConnectionTest extends TestCase
{
/**
* @expectedException \InvalidArgumentException
* @expectedExceptionMessage The given AMQP DSN "amqp://" is invalid.
* @expectedExceptionMessage The given AMQP DSN "amqp://:" is invalid.
*/
public function testItCannotBeConstructedWithAWrongDsn()
{
Connection::fromDsn('amqp://');
Connection::fromDsn('amqp://:');
}
public function testItGetsParametersFromTheDsn()
public function testItCanBeConstructedWithDefaults()
{
$this->assertEquals(
new Connection([
@ -44,7 +44,23 @@ class ConnectionTest extends TestCase
], [
'messages' => [],
]),
Connection::fromDsn('amqp://localhost/%2f/messages')
Connection::fromDsn('amqp://')
);
}
public function testItGetsParametersFromTheDsn()
{
$this->assertEquals(
new Connection([
'host' => 'host',
'port' => 5672,
'vhost' => '/',
], [
'name' => 'custom',
], [
'custom' => [],
]),
Connection::fromDsn('amqp://host/%2f/custom')
);
}
@ -52,9 +68,9 @@ class ConnectionTest extends TestCase
{
$this->assertEquals(
new Connection([
'host' => 'redis',
'host' => 'localhost',
'port' => 1234,
'vhost' => '/',
'vhost' => 'vhost',
'login' => 'guest',
'password' => 'password',
], [
@ -62,7 +78,7 @@ class ConnectionTest extends TestCase
], [
'queueName' => [],
]),
Connection::fromDsn('amqp://guest:password@redis:1234/%2f/queue?exchange[name]=exchangeName&queues[queueName]')
Connection::fromDsn('amqp://guest:password@localhost:1234/vhost/queue?exchange[name]=exchangeName&queues[queueName]')
);
}
@ -70,18 +86,16 @@ class ConnectionTest extends TestCase
{
$this->assertEquals(
new Connection([
'host' => 'redis',
'port' => 1234,
'host' => 'localhost',
'port' => 5672,
'vhost' => '/',
'login' => 'guest',
'password' => 'password',
'persistent' => 'true',
], [
'name' => 'exchangeName',
], [
'queueName' => [],
]),
Connection::fromDsn('amqp://guest:password@redis:1234/%2f/queue?exchange[name]=exchangeName&queues[queueName]', [
Connection::fromDsn('amqp://localhost/%2f/queue?exchange[name]=exchangeName&queues[queueName]', [
'persistent' => 'true',
'exchange' => ['name' => 'toBeOverwritten'],
])

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

@ -58,8 +58,22 @@ class Connection
*/
private $amqpDelayExchange;
public function __construct(array $connectionOptions, array $exchangeOptions, array $queuesOptions, AmqpFactory $amqpFactory = null)
{
$this->connectionOptions = array_replace_recursive([
'delay' => [
'routing_key_pattern' => 'delay_%routing_key%_%delay%',
'exchange_name' => 'delay',
'queue_name_pattern' => 'delay_queue_%routing_key%_%delay%',
],
], $connectionOptions);
$this->exchangeOptions = $exchangeOptions;
$this->queuesOptions = $queuesOptions;
$this->amqpFactory = $amqpFactory ?: new AmqpFactory();
}
/**
* Constructor.
* Creates a connection based on the DSN and options.
*
* Available options:
*
@ -81,29 +95,19 @@ class Connection
* * delay:
* * routing_key_pattern: The pattern of the routing key (Default: "delay_%routing_key%_%delay%")
* * queue_name_pattern: Pattern to use to create the queues (Default: "delay_queue_%routing_key%_%delay%")
* * exchange_name: Name of the exchange to be used for the retried messages (Default: "retry")
* * exchange_name: Name of the exchange to be used for the retried messages (Default: "delay")
* * auto_setup: Enable or not the auto-setup of queues and exchanges (Default: true)
* * loop_sleep: Amount of micro-seconds to wait if no message are available (Default: 200000)
* * prefetch_count: set channel prefetch count
*/
public function __construct(array $connectionOptions, array $exchangeOptions, array $queuesOptions, AmqpFactory $amqpFactory = null)
{
$this->connectionOptions = array_replace_recursive([
'delay' => [
'routing_key_pattern' => 'delay_%routing_key%_%delay%',
'exchange_name' => 'delay',
'queue_name_pattern' => 'delay_queue_%routing_key%_%delay%',
],
], $connectionOptions);
$this->exchangeOptions = $exchangeOptions;
$this->queuesOptions = $queuesOptions;
$this->amqpFactory = $amqpFactory ?: new AmqpFactory();
}
public static function fromDsn(string $dsn, array $options = [], AmqpFactory $amqpFactory = null): self
{
if (false === $parsedUrl = parse_url($dsn)) {
throw new InvalidArgumentException(sprintf('The given AMQP DSN "%s" is invalid.', $dsn));
// this is a valid URI that parse_url cannot handle when you want to pass all parameters as options
if ('amqp://' !== $dsn) {
throw new InvalidArgumentException(sprintf('The given AMQP DSN "%s" is invalid.', $dsn));
}
$parsedUrl = [];
}
$pathParts = isset($parsedUrl['path']) ? explode('/', trim($parsedUrl['path'], '/')) : [];
@ -275,18 +279,17 @@ class Connection
$queue = $this->amqpFactory->createQueue($this->channel());
$queue->setName(str_replace(
['%delay%', '%routing_key%'],
[$delay, $routingKey ?: ''],
[$delay, $routingKey ?? ''],
$this->connectionOptions['delay']['queue_name_pattern']
));
));
$queue->setArguments([
'x-message-ttl' => $delay,
'x-dead-letter-exchange' => $this->exchange()->getName(),
]);
if (null !== $routingKey) {
// after being released from to DLX, this routing key will be used
$queue->setArgument('x-dead-letter-routing-key', $routingKey);
}
// 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
$queue->setArgument('x-dead-letter-routing-key', $routingKey ?? '');
return $queue;
}
@ -295,7 +298,7 @@ class Connection
{
return str_replace(
['%delay%', '%routing_key%'],
[$delay, $finalRoutingKey ?: ''],
[$delay, $finalRoutingKey ?? ''],
$this->connectionOptions['delay']['routing_key_pattern']
);
}