vendor/symfony/messenger/DependencyInjection/MessengerPass.php line 206

  1. <?php
  3. /*
  4.  * This file is part of the Symfony package.
  5.  *
  6.  * (c) Fabien Potencier <fabien@symfony.com>
  7.  *
  8.  * For the full copyright and license information, please view the LICENSE
  9.  * file that was distributed with this source code.
  10.  */
  12. namespace Symfony\Component\Messenger\DependencyInjection;
  14. use Symfony\Component\DependencyInjection\Argument\IteratorArgument;
  15. use Symfony\Component\DependencyInjection\ChildDefinition;
  16. use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
  17. use Symfony\Component\DependencyInjection\Compiler\ServiceLocatorTagPass;
  18. use Symfony\Component\DependencyInjection\ContainerBuilder;
  19. use Symfony\Component\DependencyInjection\Definition;
  20. use Symfony\Component\DependencyInjection\Exception\OutOfBoundsException;
  21. use Symfony\Component\DependencyInjection\Exception\RuntimeException;
  22. use Symfony\Component\DependencyInjection\Reference;
  23. use Symfony\Component\Messenger\Attribute\AsMessageHandler;
  24. use Symfony\Component\Messenger\Handler\HandlerDescriptor;
  25. use Symfony\Component\Messenger\Handler\HandlersLocator;
  26. use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
  27. use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
  28. use Symfony\Component\Messenger\TraceableMessageBus;
  29. use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
  31. /**
  32.  * @author Samuel Roze <samuel.roze@gmail.com>
  33.  */
  34. class MessengerPass implements CompilerPassInterface
  35. {
  36.     public function process(ContainerBuilder $container)
  37.     {
  38.         $busIds = [];
  39.         foreach ($container->findTaggedServiceIds('messenger.bus') as $busId => $tags) {
  40.             $busIds[] = $busId;
  41.             if ($container->hasParameter($busMiddlewareParameter = $busId.'.middleware')) {
  42.                 $this->registerBusMiddleware($container, $busId, $container->getParameter($busMiddlewareParameter));
  44.                 $container->getParameterBag()->remove($busMiddlewareParameter);
  45.             }
  47.             if ($container->hasDefinition('data_collector.messenger')) {
  48.                 $this->registerBusToCollector($container, $busId);
  49.             }
  50.         }
  52.         if ($container->hasDefinition('messenger.receiver_locator')) {
  53.             $this->registerReceivers($container, $busIds);
  54.         }
  55.         $this->registerHandlers($container, $busIds);
  56.     }
  58.     private function registerHandlers(ContainerBuilder $container, array $busIds)
  59.     {
  60.         $definitions = [];
  61.         $handlersByBusAndMessage = [];
  62.         $handlerToOriginalServiceIdMapping = [];
  64.         foreach ($container->findTaggedServiceIds('messenger.message_handler', true) as $serviceId => $tags) {
  65.             foreach ($tags as $tag) {
  66.                 if (isset($tag['bus']) && !\in_array($tag['bus'], $busIds, true)) {
  67.                     throw new RuntimeException(sprintf('Invalid handler service "%s": bus "%s" specified on the tag "messenger.message_handler" does not exist (known ones are: "%s").', $serviceId, $tag['bus'], implode('", "', $busIds)));
  68.                 }
  70.                 $className = $this->getServiceClass($container, $serviceId);
  71.                 $r = $container->getReflectionClass($className);
  73.                 if (null === $r) {
  74.                     throw new RuntimeException(sprintf('Invalid service "%s": class "%s" does not exist.', $serviceId, $className));
  75.                 }
  77.                 if (isset($tag['handles'])) {
  78.                     $handles = isset($tag['method']) ? [$tag['handles'] => $tag['method']] : [$tag['handles']];
  79.                 } else {
  80.                     $handles = $this->guessHandledClasses($r, $serviceId, $tag['method'] ?? '__invoke');
  81.                 }
  83.                 $message = null;
  84.                 $handlerBuses = (array) ($tag['bus'] ?? $busIds);
  86.                 foreach ($handles as $message => $options) {
  87.                     $buses = $handlerBuses;
  89.                     if (\is_int($message)) {
  90.                         if (\is_string($options)) {
  91.                             $message = $options;
  92.                             $options = [];
  93.                         } else {
  94.                             throw new RuntimeException(sprintf('The handler configuration needs to return an array of messages or an associated array of message and configuration. Found value of type "%s" at position "%d" for service "%s".', get_debug_type($options), $message, $serviceId));
  95.                         }
  96.                     }
  98.                     if (\is_string($options)) {
  99.                         $options = ['method' => $options];
  100.                     }
  102.                     if (!isset($options['from_transport']) && isset($tag['from_transport'])) {
  103.                         $options['from_transport'] = $tag['from_transport'];
  104.                     }
  106.                     $priority = $tag['priority'] ?? $options['priority'] ?? 0;
  107.                     $method = $options['method'] ?? '__invoke';
  109.                     if (isset($options['bus'])) {
  110.                         if (!\in_array($options['bus'], $busIds)) {
  111.                             // @deprecated since Symfony 6.2, in 7.0 change to:
  112.                             // $messageLocation = isset($tag['handles']) ? 'declared in your tag attribute "handles"' : sprintf('used as argument type in method "%s::%s()"', $r->getName(), $method);
  113.                             $messageLocation = isset($tag['handles']) ? 'declared in your tag attribute "handles"' : ($r->implementsInterface(MessageSubscriberInterface::class) ? sprintf('returned by method "%s::getHandledMessages()"', $r->getName()) : sprintf('used as argument type in method "%s::%s()"', $r->getName(), $method));
  115.                             throw new RuntimeException(sprintf('Invalid configuration "%s" for message "%s": bus "%s" does not exist.', $messageLocation, $message, $options['bus']));
  116.                         }
  118.                         $buses = [$options['bus']];
  119.                     }
  121.                     if ('*' !== $message && !class_exists($message) && !interface_exists($message, false)) {
  122.                         // @deprecated since Symfony 6.2, in 7.0 change to:
  123.                         // $messageLocation = isset($tag['handles']) ? 'declared in your tag attribute "handles"' : sprintf('used as argument type in method "%s::%s()"', $r->getName(), $method);
  124.                         $messageLocation = isset($tag['handles']) ? 'declared in your tag attribute "handles"' : ($r->implementsInterface(MessageSubscriberInterface::class) ? sprintf('returned by method "%s::getHandledMessages()"', $r->getName()) : sprintf('used as argument type in method "%s::%s()"', $r->getName(), $method));
  126.                         throw new RuntimeException(sprintf('Invalid handler service "%s": class or interface "%s" "%s" not found.', $serviceId, $message, $messageLocation));
  127.                     }
  129.                     if (!$r->hasMethod($method)) {
  130.                         throw new RuntimeException(sprintf('Invalid handler service "%s": method "%s::%s()" does not exist.', $serviceId, $r->getName(), $method));
  131.                     }
  133.                     if ('__invoke' !== $method) {
  134.                         $wrapperDefinition = (new Definition('callable'))->addArgument([new Reference($serviceId), $method])->setFactory('Closure::fromCallable');
  136.                         $definitions[$definitionId = '.messenger.method_on_object_wrapper.'.ContainerBuilder::hash($message.':'.$priority.':'.$serviceId.':'.$method)] = $wrapperDefinition;
  137.                     } else {
  138.                         $definitionId = $serviceId;
  139.                     }
  141.                     $handlerToOriginalServiceIdMapping[$definitionId] = $serviceId;
  143.                     foreach ($buses as $handlerBus) {
  144.                         $handlersByBusAndMessage[$handlerBus][$message][$priority][] = [$definitionId, $options];
  145.                     }
  146.                 }
  148.                 if (null === $message) {
  149.                     throw new RuntimeException(sprintf('Invalid handler service "%s": method "%s::getHandledMessages()" must return one or more messages.', $serviceId, $r->getName()));
  150.                 }
  151.             }
  152.         }
  154.         foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
  155.             foreach ($handlersByMessage as $message => $handlersByPriority) {
  156.                 krsort($handlersByPriority);
  157.                 $handlersByBusAndMessage[$bus][$message] = array_merge(...$handlersByPriority);
  158.             }
  159.         }
  161.         $handlersLocatorMappingByBus = [];
  162.         foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
  163.             foreach ($handlersByMessage as $message => $handlers) {
  164.                 $handlerDescriptors = [];
  165.                 foreach ($handlers as $handler) {
  166.                     $definitions[$definitionId = '.messenger.handler_descriptor.'.ContainerBuilder::hash($bus.':'.$message.':'.$handler[0])] = (new Definition(HandlerDescriptor::class))->setArguments([new Reference($handler[0]), $handler[1]]);
  167.                     $handlerDescriptors[] = new Reference($definitionId);
  168.                 }
  170.                 $handlersLocatorMappingByBus[$bus][$message] = new IteratorArgument($handlerDescriptors);
  171.             }
  172.         }
  173.         $container->addDefinitions($definitions);
  175.         foreach ($busIds as $bus) {
  176.             $container->register($locatorId = $bus.'.messenger.handlers_locator', HandlersLocator::class)
  177.                 ->setArgument(0, $handlersLocatorMappingByBus[$bus] ?? [])
  178.             ;
  179.             if ($container->has($handleMessageId = $bus.'.middleware.handle_message')) {
  180.                 $container->getDefinition($handleMessageId)
  181.                     ->replaceArgument(0, new Reference($locatorId))
  182.                 ;
  183.             }
  184.         }
  186.         if ($container->hasDefinition('console.command.messenger_debug')) {
  187.             $debugCommandMapping = $handlersByBusAndMessage;
  188.             foreach ($busIds as $bus) {
  189.                 if (!isset($debugCommandMapping[$bus])) {
  190.                     $debugCommandMapping[$bus] = [];
  191.                 }
  193.                 foreach ($debugCommandMapping[$bus] as $message => $handlers) {
  194.                     foreach ($handlers as $key => $handler) {
  195.                         $debugCommandMapping[$bus][$message][$key][0] = $handlerToOriginalServiceIdMapping[$handler[0]];
  196.                     }
  197.                 }
  198.             }
  199.             $container->getDefinition('console.command.messenger_debug')->replaceArgument(0, $debugCommandMapping);
  200.         }
  201.     }
  203.     private function guessHandledClasses(\ReflectionClass $handlerClass, string $serviceId, string $methodName): iterable
  204.     {
  205.         if ($handlerClass->implementsInterface(MessageSubscriberInterface::class)) {
  206.             trigger_deprecation('symfony/messenger', '6.2', 'Implementing "%s" is deprecated, use the "%s" attribute instead.', MessageSubscriberInterface::class, AsMessageHandler::class);
  208.             return $handlerClass->getName()::getHandledMessages();
  209.         }
  211.         if ($handlerClass->implementsInterface(MessageHandlerInterface::class)) {
  212.             trigger_deprecation('symfony/messenger', '6.2', 'Implementing "%s" is deprecated, use the "%s" attribute instead.', MessageHandlerInterface::class, AsMessageHandler::class);
  213.         }
  215.         try {
  216.             $method = $handlerClass->getMethod($methodName);
  217.         } catch (\ReflectionException) {
  218.             throw new RuntimeException(sprintf('Invalid handler service "%s": class "%s" must have an "%s()" method.', $serviceId, $handlerClass->getName(), $methodName));
  219.         }
  221.         if (0 === $method->getNumberOfRequiredParameters()) {
  222.             throw new RuntimeException(sprintf('Invalid handler service "%s": method "%s::%s()" requires at least one argument, first one being the message it handles.', $serviceId, $handlerClass->getName(), $methodName));
  223.         }
  225.         $parameters = $method->getParameters();
  227.         /** @var \ReflectionNamedType|\ReflectionUnionType|null */
  228.         $type = $parameters[0]->getType();
  230.         if (!$type) {
  231.             throw new RuntimeException(sprintf('Invalid handler service "%s": argument "$%s" of method "%s::%s()" must have a type-hint corresponding to the message class it handles.', $serviceId, $parameters[0]->getName(), $handlerClass->getName(), $methodName));
  232.         }
  234.         if ($type instanceof \ReflectionUnionType) {
  235.             $types = [];
  236.             $invalidTypes = [];
  237.             foreach ($type->getTypes() as $type) {
  238.                 if (!$type->isBuiltin()) {
  239.                     $types[] = (string) $type;
  240.                 } else {
  241.                     $invalidTypes[] = (string) $type;
  242.                 }
  243.             }
  245.             if ($types) {
  246.                 return ('__invoke' === $methodName) ? $types : array_fill_keys($types, $methodName);
  247.             }
  249.             throw new RuntimeException(sprintf('Invalid handler service "%s": type-hint of argument "$%s" in method "%s::__invoke()" must be a class , "%s" given.', $serviceId, $parameters[0]->getName(), $handlerClass->getName(), implode('|', $invalidTypes)));
  250.         }
  252.         if ($type->isBuiltin()) {
  253.             throw new RuntimeException(sprintf('Invalid handler service "%s": type-hint of argument "$%s" in method "%s::%s()" must be a class , "%s" given.', $serviceId, $parameters[0]->getName(), $handlerClass->getName(), $methodName, $type instanceof \ReflectionNamedType ? $type->getName() : (string) $type));
  254.         }
  256.         return ('__invoke' === $methodName) ? [$type->getName()] : [$type->getName() => $methodName];
  257.     }
  259.     private function registerReceivers(ContainerBuilder $container, array $busIds)
  260.     {
  261.         $receiverMapping = [];
  262.         $failureTransportsMap = [];
  263.         if ($container->hasDefinition('console.command.messenger_failed_messages_retry')) {
  264.             $commandDefinition = $container->getDefinition('console.command.messenger_failed_messages_retry');
  265.             $globalReceiverName = $commandDefinition->getArgument(0);
  266.             if (null !== $globalReceiverName) {
  267.                 if ($container->hasAlias('messenger.failure_transports.default')) {
  268.                     $failureTransportsMap[$globalReceiverName] = new Reference('messenger.failure_transports.default');
  269.                 } else {
  270.                     $failureTransportsMap[$globalReceiverName] = new Reference('messenger.transport.'.$globalReceiverName);
  271.                 }
  272.             }
  273.         }
  275.         foreach ($container->findTaggedServiceIds('messenger.receiver') as $id => $tags) {
  276.             $receiverClass = $this->getServiceClass($container, $id);
  277.             if (!is_subclass_of($receiverClass, ReceiverInterface::class)) {
  278.                 throw new RuntimeException(sprintf('Invalid receiver "%s": class "%s" must implement interface "%s".', $id, $receiverClass, ReceiverInterface::class));
  279.             }
  281.             $receiverMapping[$id] = new Reference($id);
  283.             foreach ($tags as $tag) {
  284.                 if (isset($tag['alias'])) {
  285.                     $receiverMapping[$tag['alias']] = $receiverMapping[$id];
  286.                     if ($tag['is_failure_transport'] ?? false) {
  287.                         $failureTransportsMap[$tag['alias']] = $receiverMapping[$id];
  288.                     }
  289.                 }
  290.             }
  291.         }
  293.         $receiverNames = [];
  294.         foreach ($receiverMapping as $name => $reference) {
  295.             $receiverNames[(string) $reference] = $name;
  296.         }
  298.         $buses = [];
  299.         foreach ($busIds as $busId) {
  300.             $buses[$busId] = new Reference($busId);
  301.         }
  303.         if ($hasRoutableMessageBus = $container->hasDefinition('messenger.routable_message_bus')) {
  304.             $container->getDefinition('messenger.routable_message_bus')
  305.                 ->replaceArgument(0, ServiceLocatorTagPass::register($container, $buses));
  306.         }
  308.         if ($container->hasDefinition('console.command.messenger_consume_messages')) {
  309.             $consumeCommandDefinition = $container->getDefinition('console.command.messenger_consume_messages');
  311.             if ($hasRoutableMessageBus) {
  312.                 $consumeCommandDefinition->replaceArgument(0, new Reference('messenger.routable_message_bus'));
  313.             }
  315.             $consumeCommandDefinition->replaceArgument(4, array_values($receiverNames));
  316.             try {
  317.                 $consumeCommandDefinition->replaceArgument(6, $busIds);
  318.             } catch (OutOfBoundsException) {
  319.                 // ignore to preserve compatibility with symfony/framework-bundle < 5.4
  320.             }
  321.         }
  323.         if ($container->hasDefinition('console.command.messenger_setup_transports')) {
  324.             $container->getDefinition('console.command.messenger_setup_transports')
  325.                 ->replaceArgument(1, array_values($receiverNames));
  326.         }
  328.         if ($container->hasDefinition('console.command.messenger_stats')) {
  329.             $container->getDefinition('console.command.messenger_stats')
  330.                 ->replaceArgument(1, array_values($receiverNames));
  331.         }
  333.         $container->getDefinition('messenger.receiver_locator')->replaceArgument(0, $receiverMapping);
  335.         $failureTransportsLocator = ServiceLocatorTagPass::register($container, $failureTransportsMap);
  337.         $failedCommandIds = [
  338.             'console.command.messenger_failed_messages_retry',
  339.             'console.command.messenger_failed_messages_show',
  340.             'console.command.messenger_failed_messages_remove',
  341.         ];
  342.         foreach ($failedCommandIds as $failedCommandId) {
  343.             if ($container->hasDefinition($failedCommandId)) {
  344.                 $definition = $container->getDefinition($failedCommandId);
  345.                 $definition->replaceArgument(1, $failureTransportsLocator);
  346.             }
  347.         }
  348.     }
  350.     private function registerBusToCollector(ContainerBuilder $container, string $busId)
  351.     {
  352.         $container->setDefinition(
  353.             $tracedBusId = 'debug.traced.'.$busId,
  354.             (new Definition(TraceableMessageBus::class, [new Reference($tracedBusId.'.inner')]))->setDecoratedService($busId)
  355.         );
  357.         $container->getDefinition('data_collector.messenger')->addMethodCall('registerBus', [$busId, new Reference($tracedBusId)]);
  358.     }
  360.     private function registerBusMiddleware(ContainerBuilder $container, string $busId, array $middlewareCollection)
  361.     {
  362.         $middlewareReferences = [];
  363.         foreach ($middlewareCollection as $middlewareItem) {
  364.             $id = $middlewareItem['id'];
  365.             $arguments = $middlewareItem['arguments'] ?? [];
  366.             if (!$container->has($messengerMiddlewareId = 'messenger.middleware.'.$id)) {
  367.                 $messengerMiddlewareId = $id;
  368.             }
  370.             if (!$container->has($messengerMiddlewareId)) {
  371.                 throw new RuntimeException(sprintf('Invalid middleware: service "%s" not found.', $id));
  372.             }
  374.             if ($container->findDefinition($messengerMiddlewareId)->isAbstract()) {
  375.                 $childDefinition = new ChildDefinition($messengerMiddlewareId);
  376.                 $childDefinition->setArguments($arguments);
  377.                 if (isset($middlewareReferences[$messengerMiddlewareId = $busId.'.middleware.'.$id])) {
  378.                     $messengerMiddlewareId .= '.'.ContainerBuilder::hash($arguments);
  379.                 }
  380.                 $container->setDefinition($messengerMiddlewareId, $childDefinition);
  381.             } elseif ($arguments) {
  382.                 throw new RuntimeException(sprintf('Invalid middleware factory "%s": a middleware factory must be an abstract definition.', $id));
  383.             }
  385.             $middlewareReferences[$messengerMiddlewareId] = new Reference($messengerMiddlewareId);
  386.         }
  388.         $container->getDefinition($busId)->replaceArgument(0, new IteratorArgument(array_values($middlewareReferences)));
  389.     }
  391.     private function getServiceClass(ContainerBuilder $container, string $serviceId): string
  392.     {
  393.         while (true) {
  394.             $definition = $container->findDefinition($serviceId);
  396.             if (!$definition->getClass() && $definition instanceof ChildDefinition) {
  397.                 $serviceId = $definition->getParent();
  399.                 continue;
  400.             }
  402.             return $definition->getClass();
  403.         }
  404.     }
  405. }