Skip to content

Commit

Permalink
bug #32035 [Messenger] fix delay delivery for non-fanout exchanges (T…
Browse files Browse the repository at this point in the history 
…obion)

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
-------

0f15306 [Messenger] fix delay delivery for non-fanout exchanges
  • Loading branch information
@Tobion
Tobion committed Jun 14, 2019
2 parents 6b50c89 + 0f15306 commit a9bcdcc
Show file tree
Hide file tree
Showing 2 changed files with 54 additions and 37 deletions.
38 changes: 26 additions & 12 deletions src/Symfony/Component/Messenger/Tests/Transport/AmqpExt/ConnectionTest.php
View file
Expand Up @@ -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([
Expand All @@ -44,44 +44,58 @@ public function testItGetsParametersFromTheDsn()
], [
'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')
);
}

public function testOverrideOptionsViaQueryParameters()
{
$this->assertEquals(
new Connection([
'host' => 'redis',
'host' => 'localhost',
'port' => 1234,
'vhost' => '/',
'vhost' => 'vhost',
'login' => 'guest',
'password' => 'password',
], [
'name' => 'exchangeName',
], [
'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]')
);
}

public function testOptionsAreTakenIntoAccountAndOverwrittenByDsn()
{
$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'],
])
Expand Down
53 changes: 28 additions & 25 deletions src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php
View file
Expand Up @@ -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:
*
Expand All @@ -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'], '/')) : [];
Expand Down Expand Up @@ -275,18 +279,17 @@ private function createDelayQueue(int $delay, ?string $routingKey)
$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;
}
Expand All @@ -295,7 +298,7 @@ private function getRoutingKeyForDelay(int $delay, ?string $finalRoutingKey): st
{
return str_replace(
['%delay%', '%routing_key%'],
[$delay, $finalRoutingKey ?: ''],
[$delay, $finalRoutingKey ?? ''],
$this->connectionOptions['delay']['routing_key_pattern']
);
}
Expand Down

0 comments on commit a9bcdcc

Please sign in to comment.