[Messenger] Add simple transport based rate limiter to Messenger

Author: bobvandevijver

CHANGELOG.md

@@ -15,6 +15,7 @@ CHANGELOG
  * Tag all workflows services with `workflow`, those with type=workflow are
    tagged with `workflow.workflow`, and those with type=state_machine with
    `workflow.state_machine`
+ * Add `rate_limiter` configuration option to `messenger.transport` to allow rate limited transports using the RateLimiter component
 
 6.1
 ---

DependencyInjection/Configuration.php

@@ -1492,6 +1492,10 @@ function ($a) {
                                             ->integerNode('max_delay')->defaultValue(0)->min(0)->info('Max time in ms that a retry should ever be delayed (0 = infinite)')->end()
                                         ->end()
                                     ->end()
+                                    ->scalarNode('rate_limiter')
+                                        ->defaultNull()
+                                        ->info('Rate limiter name to use when processing messages')
+                                    ->end()
                                 ->end()
                             ->end()
                         ->end()

DependencyInjection/FrameworkExtension.php

@@ -2102,6 +2102,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
 
         $senderAliases = [];
         $transportRetryReferences = [];
+        $transportRateLimiterReferences = [];
         foreach ($config['transports'] as $name => $transport) {
             $serializerId = $transport['serializer'] ?? 'messenger.default_serializer';
             $transportDefinition = (new Definition(TransportInterface::class))
@@ -2130,6 +2131,14 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
 
                 $transportRetryReferences[$name] = new Reference($retryServiceId);
             }
+
+            if ($transport['rate_limiter']) {
+                if (!interface_exists(LimiterInterface::class)) {
+                    throw new LogicException('Rate limiter cannot be used within Messenger as the RateLimiter component is not installed. Try running "composer require symfony/rate-limiter".');
+                }
+
+                $transportRateLimiterReferences[$name] = new Reference('limiter.'.$transport['rate_limiter']);
+            }
         }
 
         $senderReferences = [];
@@ -2184,6 +2193,13 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
         $container->getDefinition('messenger.retry_strategy_locator')
             ->replaceArgument(0, $transportRetryReferences);
 
+        if (!$transportRateLimiterReferences) {
+            $container->removeDefinition('messenger.rate_limiter_locator');
+        } else {
+            $container->getDefinition('messenger.rate_limiter_locator')
+                ->replaceArgument(0, $transportRateLimiterReferences);
+        }
+
         if (\count($failureTransports) > 0) {
             $container->getDefinition('console.command.messenger_failed_messages_retry')
                 ->replaceArgument(0, $config['failure_transport']);

Resources/config/console.php

@@ -160,6 +160,7 @@
                 [], // Receiver names
                 service('messenger.listener.reset_services')->nullOnInvalid(),
                 [], // Bus names
+                service('messenger.rate_limiter_locator')->nullOnInvalid(),
             ])
             ->tag('console.command')
             ->tag('monolog.logger', ['channel' => 'messenger'])

Resources/config/messenger.php

@@ -159,6 +159,11 @@
                 abstract_arg('max delay ms'),
             ])
 
+        // rate limiter
+        ->set('messenger.rate_limiter_locator', ServiceLocator::class)
+            ->args([[]])
+            ->tag('container.service_locator')
+
         // worker event listener
         ->set('messenger.retry.send_failed_message_for_retry_listener', SendFailedMessageForRetryListener::class)
             ->args([

Resources/config/schema/symfony-1.0.xsd

@@ -554,6 +554,7 @@
         <xsd:attribute name="serializer" type="xsd:string" />
         <xsd:attribute name="dsn" type="xsd:string" />
         <xsd:attribute name="failure-transport" type="xsd:string" />
+        <xsd:attribute name="rate-limiter" type="xsd:string" />
     </xsd:complexType>
 
     <xsd:complexType name="messenger_retry_strategy">

Tests/DependencyInjection/Fixtures/php/messenger_transports.php

@@ -20,6 +20,7 @@
                     'multiplier' => 3,
                     'max_delay' => 100,
                 ],
+                'rate_limiter' => 'customised_worker'
             ],
             'failed' => 'in-memory:///',
             'redis' => 'redis://127.0.0.1:6379/messages',

Tests/DependencyInjection/Fixtures/xml/messenger_transports.xml

@@ -10,7 +10,7 @@
         <framework:messenger failure-transport="failed">
             <framework:serializer default-serializer="messenger.transport.symfony_serializer" />
             <framework:transport name="default" dsn="amqp://localhost/%2f/messages" />
-            <framework:transport name="customised" dsn="amqp://localhost/%2f/messages?exchange_name=exchange_name" serializer="messenger.transport.native_php_serializer">
+            <framework:transport name="customised" dsn="amqp://localhost/%2f/messages?exchange_name=exchange_name" serializer="messenger.transport.native_php_serializer" rate-limiter="customised_worker">
                 <framework:options>
                     <framework:queue>
                         <framework:name>Queue</framework:name>

Tests/DependencyInjection/Fixtures/yml/messenger_transports.yml

@@ -18,6 +18,7 @@ framework:
                     delay: 7
                     multiplier: 3
                     max_delay: 100
+                rate_limiter: customised_worker
             failed: 'in-memory:///'
             redis: 'redis://127.0.0.1:6379/messages'
             beanstalkd: 'beanstalkd://127.0.0.1:11300'

Tests/DependencyInjection/FrameworkExtensionTest.php

@@ -949,6 +949,12 @@ public function testMessengerTransports()
             return array_shift($values);
         }, $failureTransports);
         $this->assertEquals($expectedTransportsByFailureTransports, $failureTransportsReferences);
+
+        $rateLimitedTransports = $container->getDefinition('messenger.rate_limiter_locator')->getArgument(0);
+        $expectedRateLimitersByRateLimitedTransports = [
+            'customised' => new Reference('limiter.customised_worker'),
+        ];
+        $this->assertEquals($expectedRateLimitersByRateLimitedTransports, $rateLimitedTransports);
     }
 
     public function testMessengerRouting()