Skip to content

[Messenger] Prioritize receivers via transport configuration #57757

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 1 commit into
base: 7.4
Choose a base branch
from
Open

[Messenger] Prioritize receivers via transport configuration #57757

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
1 change: 1 addition & 0 deletions src/Symfony/Bundle/FrameworkBundle/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ CHANGELOG
* Make the `config/` directory optional in `MicroKernelTrait`, add support for service arguments in the
invokable Kernel class, and register `FrameworkBundle` by default when the `bundles.php` file is missing
* [BC BREAK] The `secrets:decrypt-to-local` command terminates with a non-zero exit code when a secret could not be read
* Add `priority` option to the `messenger.transports` configurations and pass the configured priority to the `messenger.receiver` service tag

7.1
---
Expand Down
1 change: 1 addition & 0 deletions src/Symfony/Bundle/FrameworkBundle/DependencyInjection/Configuration.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1612,6 +1612,7 @@ function ($a) {
->defaultNull()
->info('Rate limiter name to use when processing messages')
->end()
->integerNode('priority')->defaultValue(0)->info('Priority of this transport when the consumer is executed with the --all flag')->end()
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
->integerNode('priority')->defaultValue(0)->info('Priority of this transport when the consumer is executed with the --all flag')->end()
->integerNode('priority')
->defaultValue(0)
->info('Priority of this transport when the consumer is executed with the --all flag')
->end()

->end()
->end()
->end()
Expand Down
1 change: 1 addition & 0 deletions src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2220,6 +2220,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
->addTag('messenger.receiver', [
'alias' => $name,
'is_failure_transport' => \in_array($name, $failureTransports, true),
'priority' => $transport['priority'],
])
;
$container->setDefinition($transportId = 'messenger.transport.'.$name, $transportDefinition);
Expand Down
1 change: 1 addition & 0 deletions src/Symfony/Bundle/FrameworkBundle/Resources/config/schema/symfony-1.0.xsd
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -606,6 +606,7 @@
<xsd:attribute name="dsn" type="xsd:string" />
<xsd:attribute name="failure-transport" type="xsd:string" />
<xsd:attribute name="rate-limiter" type="xsd:string" />
<xsd:attribute name="priority" type="xsd:integer" />
</xsd:complexType>

<xsd:complexType name="messenger_retry_strategy">
Expand Down
4 changes: 4 additions & 0 deletions ...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/php/messenger_transports.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,10 @@
],
'rate_limiter' => 'customised_worker'
],
'prioritized' => [
'dsn' => 'amqp://localhost/%2f/messages?exchange_name=priority',
'priority' => 10,
],
'failed' => 'in-memory:///',
'redis' => 'redis://127.0.0.1:6379/messages',
'beanstalkd' => 'beanstalkd://127.0.0.1:11300',
Expand Down
1 change: 1 addition & 0 deletions ...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/xml/messenger_transports.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
</framework:options>
<framework:retry-strategy max-retries="10" delay="7" multiplier="3" max-delay="100"/>
</framework:transport>
<framework:transport name="prioritized" dsn="amqp://localhost/%2f/messages?exchange_name=priority" priority="10" />
<framework:transport name="failed" dsn="in-memory:///" />
<framework:transport name="redis" dsn="redis://127.0.0.1:6379/messages" />
<framework:transport name="beanstalkd" dsn="beanstalkd://127.0.0.1:11300" />
Expand Down
3 changes: 3 additions & 0 deletions ...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/yml/messenger_transports.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,9 @@ framework:
multiplier: 3
max_delay: 100
rate_limiter: customised_worker
prioritized:
dsn: amqp://localhost/%2f/messages?exchange_name=priority
priority: 10
failed: 'in-memory:///'
redis: 'redis://127.0.0.1:6379/messages'
beanstalkd: 'beanstalkd://127.0.0.1:11300'
Expand Down
20 changes: 18 additions & 2 deletions src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTestCase.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -873,6 +873,7 @@ public function testMessengerMultipleFailureTransports()
$this->assertEquals([
'alias' => 'failure_transport_1',
'is_failure_transport' => true,
'priority' => 0,
], $failureTransport1Tags);

$failureTransport3Definition = $container->getDefinition('messenger.transport.failure_transport_3');
Expand All @@ -881,6 +882,7 @@ public function testMessengerMultipleFailureTransports()
$this->assertEquals([
'alias' => 'failure_transport_3',
'is_failure_transport' => true,
'priority' => 0,
], $failureTransport3Tags);

// transport 2 exists but does not appear in the mapping
Expand Down Expand Up @@ -913,6 +915,7 @@ public function testMessengerMultipleFailureTransportsWithGlobalFailureTransport
$this->assertEquals([
'alias' => 'failure_transport_1',
'is_failure_transport' => true,
'priority' => 0,
], $failureTransport1Tags);

$failureTransport3Definition = $container->getDefinition('messenger.transport.failure_transport_3');
Expand All @@ -921,6 +924,7 @@ public function testMessengerMultipleFailureTransportsWithGlobalFailureTransport
$this->assertEquals([
'alias' => 'failure_transport_3',
'is_failure_transport' => true,
'priority' => 0,
], $failureTransport3Tags);

$failureTransportsByTransportNameServiceLocator = $container->getDefinition('messenger.failure.send_failed_message_to_failure_transport_listener')->getArgument(0);
Expand All @@ -947,15 +951,26 @@ public function testMessengerTransports()
$container = $this->createContainerFromFile('messenger_transports');
$this->assertTrue($container->hasDefinition('messenger.transport.default'));
$this->assertTrue($container->getDefinition('messenger.transport.default')->hasTag('messenger.receiver'));
$this->assertEquals([
['alias' => 'default', 'is_failure_transport' => false], ], $container->getDefinition('messenger.transport.default')->getTag('messenger.receiver'));
$this->assertEquals([[
'alias' => 'default',
'is_failure_transport' => false,
'priority' => 0,
]], $container->getDefinition('messenger.transport.default')->getTag('messenger.receiver'));
$transportArguments = $container->getDefinition('messenger.transport.default')->getArguments();
$this->assertEquals(new Reference('messenger.default_serializer'), $transportArguments[2]);

$this->assertTrue($container->hasDefinition('messenger.transport.customised'));
$transportFactory = $container->getDefinition('messenger.transport.customised')->getFactory();
$transportArguments = $container->getDefinition('messenger.transport.customised')->getArguments();

$this->assertTrue($container->hasDefinition('messenger.transport.prioritized'));
$this->assertTrue($container->getDefinition('messenger.transport.prioritized')->hasTag('messenger.receiver'));
$this->assertEquals([[
'alias' => 'prioritized',
'is_failure_transport' => false,
'priority' => 10,
]], $container->getDefinition('messenger.transport.prioritized')->getTag('messenger.receiver'));

$this->assertEquals([new Reference('messenger.transport_factory'), 'createTransport'], $transportFactory);
$this->assertCount(3, $transportArguments);
$this->assertSame('amqp://localhost/%2f/messages?exchange_name=exchange_name', $transportArguments[0]);
Expand Down Expand Up @@ -1000,6 +1015,7 @@ public function testMessengerTransports()
$failureTransportsByTransportNameServiceLocator = $container->getDefinition('messenger.failure.send_failed_message_to_failure_transport_listener')->getArgument(0);
$failureTransports = $container->getDefinition((string) $failureTransportsByTransportNameServiceLocator)->getArgument(0);
$expectedTransportsByFailureTransports = [
'prioritized' => new Reference('messenger.transport.failed'),
'beanstalkd' => new Reference('messenger.transport.failed'),
'customised' => new Reference('messenger.transport.failed'),
'default' => new Reference('messenger.transport.failed'),
Expand Down
1 change: 1 addition & 0 deletions src/Symfony/Component/Messenger/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ CHANGELOG
* `WrappedExceptionsInterface` now extends PHP's `Throwable` interface
* Add `#[AsMessage]` attribute with `$transport` parameter for message routing
* Add `--format` option to the `messenger:stats` command
* Allow prioritizing receivers so that `messenger:consume --all` consumes receivers in a predefined order

7.1
---
Expand Down
10 changes: 10 additions & 0 deletions src/Symfony/Component/Messenger/DependencyInjection/MessengerPass.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,6 +242,8 @@ private function registerReceivers(ContainerBuilder $container, array $busIds):
{
$receiverMapping = [];
$failureTransportsMap = [];
$receiverPriority = [];

if ($container->hasDefinition('console.command.messenger_failed_messages_retry')) {
$commandDefinition = $container->getDefinition('console.command.messenger_failed_messages_retry');
$globalReceiverName = $commandDefinition->getArgument(0);
Expand All @@ -263,15 +265,23 @@ private function registerReceivers(ContainerBuilder $container, array $busIds):
$receiverMapping[$id] = new Reference($id);

foreach ($tags as $tag) {
$receiverPriority[$id] = max($tag['priority'] ?? 0, $receiverPriority[$id] ?? PHP_INT_MIN);
Copy link
Author

@petervanderwal petervanderwal Jul 17, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

Does it happen that a receiver service has multiple messenger.receiver tags? If that never happens then the "take the max priority of all added tags" logic can be simplified to a single easier to read $receiverPriority[$id] = $tag['priority'] ?? 0.

Based on the current FrameworkExtension code it looked like the answer would be "no, that never happens" but I didn't know whether there were other edge-cases.

Copy link
Member

Choose a reason for hiding this comment

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

why not using PriorityTaggedServiceTrait::findAndSortTaggedServices instead?


if (isset($tag['alias'])) {
$receiverMapping[$tag['alias']] = $receiverMapping[$id];
$receiverPriority[$tag['alias']] = max($tag['priority'] ?? 0, $receiverPriority[$tag['alias']] ?? PHP_INT_MIN);

if ($tag['is_failure_transport'] ?? false) {
$failureTransportsMap[$tag['alias']] = $receiverMapping[$id];
}
}
}
}

$prioritySorter = fn (string $a, string $b): int => $receiverPriority[$b] <=> $receiverPriority[$a];
uksort($receiverMapping, $prioritySorter);
uksort($failureTransportsMap, $prioritySorter);

$receiverNames = [];
foreach ($receiverMapping as $name => $reference) {
$receiverNames[(string) $reference] = $name;
Expand Down
7 changes: 5 additions & 2 deletions src/Symfony/Component/Messenger/Tests/DependencyInjection/MessengerPassTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
use Symfony\Component\DependencyInjection\ServiceLocator;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpReceiver;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineReceiver;
use Symfony\Component\Messenger\Command\ConsumeMessagesCommand;
use Symfony\Component\Messenger\Command\DebugCommand;
use Symfony\Component\Messenger\Command\FailedMessagesRetryCommand;
Expand Down Expand Up @@ -448,11 +449,12 @@ public function testItRegistersMultipleReceiversAndSetsTheReceiverNamesOnTheComm
]);

$container->register(AmqpReceiver::class, AmqpReceiver::class)->addTag('messenger.receiver', ['alias' => 'amqp']);
$container->register(DoctrineReceiver::class, DoctrineReceiver::class)->addTag('messenger.receiver', ['alias' => 'doctrine', 'priority' => 1]);
$container->register(DummyReceiver::class, DummyReceiver::class)->addTag('messenger.receiver', ['alias' => 'dummy']);

(new MessengerPass())->process($container);

$this->assertSame(['amqp', 'dummy'], $container->getDefinition('console.command.messenger_consume_messages')->getArgument(4));
$this->assertSame(['doctrine', 'amqp', 'dummy'], $container->getDefinition('console.command.messenger_consume_messages')->getArgument(4));
}

public function testItSetsTheReceiverNamesOnTheSetupTransportsCommand()
Expand All @@ -464,11 +466,12 @@ public function testItSetsTheReceiverNamesOnTheSetupTransportsCommand()
]);

$container->register(AmqpReceiver::class, AmqpReceiver::class)->addTag('messenger.receiver', ['alias' => 'amqp']);
$container->register(DoctrineReceiver::class, DoctrineReceiver::class)->addTag('messenger.receiver', ['alias' => 'doctrine', 'priority' => 1]);
$container->register(DummyReceiver::class, DummyReceiver::class)->addTag('messenger.receiver', ['alias' => 'dummy']);

(new MessengerPass())->process($container);

$this->assertSame(['amqp', 'dummy'], $container->getDefinition('console.command.messenger_setup_transports')->getArgument(1));
$this->assertSame(['doctrine', 'amqp', 'dummy'], $container->getDefinition('console.command.messenger_setup_transports')->getArgument(1));
}

public function testItRegistersHandlersOnDifferentBuses()
Expand Down
Loading