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

  1. <?php
  2. /*
  3.  * This file is part of the Symfony package.
  4.  *
  5.  * (c) Fabien Potencier <fabien@symfony.com>
  6.  *
  7.  * For the full copyright and license information, please view the LICENSE
  8.  * file that was distributed with this source code.
  9.  */
  10. namespace Symfony\Component\Messenger\DependencyInjection;
  11. use Symfony\Component\DependencyInjection\Argument\IteratorArgument;
  12. use Symfony\Component\DependencyInjection\ChildDefinition;
  13. use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
  14. use Symfony\Component\DependencyInjection\Compiler\ServiceLocatorTagPass;
  15. use Symfony\Component\DependencyInjection\ContainerBuilder;
  16. use Symfony\Component\DependencyInjection\Definition;
  17. use Symfony\Component\DependencyInjection\Exception\OutOfBoundsException;
  18. use Symfony\Component\DependencyInjection\Exception\RuntimeException;
  19. use Symfony\Component\DependencyInjection\Reference;
  20. use Symfony\Component\Messenger\Attribute\AsMessageHandler;
  21. use Symfony\Component\Messenger\Handler\HandlerDescriptor;
  22. use Symfony\Component\Messenger\Handler\HandlersLocator;
  23. use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
  24. use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
  25. use Symfony\Component\Messenger\TraceableMessageBus;
  26. use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
  27. /**
  28.  * @author Samuel Roze <samuel.roze@gmail.com>
  29.  */
  30. class MessengerPass implements CompilerPassInterface
  31. {
  32.     public function process(ContainerBuilder $container)
  33.     {
  34.         $busIds = [];
  35.         foreach ($container->findTaggedServiceIds('messenger.bus') as $busId => $tags) {
  36.             $busIds[] = $busId;
  37.             if ($container->hasParameter($busMiddlewareParameter $busId.'.middleware')) {
  38.                 $this->registerBusMiddleware($container$busId$container->getParameter($busMiddlewareParameter));
  39.                 $container->getParameterBag()->remove($busMiddlewareParameter);
  40.             }
  41.             if ($container->hasDefinition('data_collector.messenger')) {
  42.                 $this->registerBusToCollector($container$busId);
  43.             }
  44.         }
  45.         if ($container->hasDefinition('messenger.receiver_locator')) {
  46.             $this->registerReceivers($container$busIds);
  47.         }
  48.         $this->registerHandlers($container$busIds);
  49.     }
  50.     private function registerHandlers(ContainerBuilder $container, array $busIds)
  51.     {
  52.         $definitions = [];
  53.         $handlersByBusAndMessage = [];
  54.         $handlerToOriginalServiceIdMapping = [];
  55.         foreach ($container->findTaggedServiceIds('messenger.message_handler'true) as $serviceId => $tags) {
  56.             foreach ($tags as $tag) {
  57.                 if (isset($tag['bus']) && !\in_array($tag['bus'], $busIdstrue)) {
  58.                     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)));
  59.                 }
  60.                 $className $this->getServiceClass($container$serviceId);
  61.                 $r $container->getReflectionClass($className);
  62.                 if (null === $r) {
  63.                     throw new RuntimeException(sprintf('Invalid service "%s": class "%s" does not exist.'$serviceId$className));
  64.                 }
  65.                 if (isset($tag['handles'])) {
  66.                     $handles = isset($tag['method']) ? [$tag['handles'] => $tag['method']] : [$tag['handles']];
  67.                 } else {
  68.                     $handles $this->guessHandledClasses($r$serviceId$tag['method'] ?? '__invoke');
  69.                 }
  70.                 $message null;
  71.                 $handlerBuses = (array) ($tag['bus'] ?? $busIds);
  72.                 foreach ($handles as $message => $options) {
  73.                     $buses $handlerBuses;
  74.                     if (\is_int($message)) {
  75.                         if (\is_string($options)) {
  76.                             $message $options;
  77.                             $options = [];
  78.                         } else {
  79.                             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));
  80.                         }
  81.                     }
  82.                     if (\is_string($options)) {
  83.                         $options = ['method' => $options];
  84.                     }
  85.                     $options += array_filter($tag);
  86.                     unset($options['handles']);
  87.                     $priority $options['priority'] ?? 0;
  88.                     $method $options['method'] ?? '__invoke';
  89.                     if (isset($options['bus'])) {
  90.                         if (!\in_array($options['bus'], $busIds)) {
  91.                             // @deprecated since Symfony 6.2, in 7.0 change to:
  92.                             // $messageLocation = isset($tag['handles']) ? 'declared in your tag attribute "handles"' : sprintf('used as argument type in method "%s::%s()"', $r->getName(), $method);
  93.                             $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));
  94.                             throw new RuntimeException(sprintf('Invalid configuration '.$messageLocation.' for message "%s": bus "%s" does not exist.'$message$options['bus']));
  95.                         }
  96.                         $buses = [$options['bus']];
  97.                     }
  98.                     if ('*' !== $message && !class_exists($message) && !interface_exists($messagefalse)) {
  99.                         // @deprecated since Symfony 6.2, in 7.0 change to:
  100.                         // $messageLocation = isset($tag['handles']) ? 'declared in your tag attribute "handles"' : sprintf('used as argument type in method "%s::%s()"', $r->getName(), $method);
  101.                         $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));
  102.                         throw new RuntimeException(sprintf('Invalid handler service "%s": class or interface "%s" '.$messageLocation.' not found.'$serviceId$message));
  103.                     }
  104.                     if (!$r->hasMethod($method)) {
  105.                         throw new RuntimeException(sprintf('Invalid handler service "%s": method "%s::%s()" does not exist.'$serviceId$r->getName(), $method));
  106.                     }
  107.                     if ('__invoke' !== $method) {
  108.                         $wrapperDefinition = (new Definition('Closure'))->addArgument([new Reference($serviceId), $method])->setFactory('Closure::fromCallable');
  109.                         $definitions[$definitionId '.messenger.method_on_object_wrapper.'.ContainerBuilder::hash($message.':'.$priority.':'.$serviceId.':'.$method)] = $wrapperDefinition;
  110.                     } else {
  111.                         $definitionId $serviceId;
  112.                     }
  113.                     $handlerToOriginalServiceIdMapping[$definitionId] = $serviceId;
  114.                     foreach ($buses as $handlerBus) {
  115.                         $handlersByBusAndMessage[$handlerBus][$message][$priority][] = [$definitionId$options];
  116.                     }
  117.                 }
  118.                 if (null === $message) {
  119.                     throw new RuntimeException(sprintf('Invalid handler service "%s": method "%s::getHandledMessages()" must return one or more messages.'$serviceId$r->getName()));
  120.                 }
  121.             }
  122.         }
  123.         foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
  124.             foreach ($handlersByMessage as $message => $handlersByPriority) {
  125.                 krsort($handlersByPriority);
  126.                 $handlersByBusAndMessage[$bus][$message] = array_merge(...$handlersByPriority);
  127.             }
  128.         }
  129.         $handlersLocatorMappingByBus = [];
  130.         foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
  131.             foreach ($handlersByMessage as $message => $handlers) {
  132.                 $handlerDescriptors = [];
  133.                 foreach ($handlers as $handler) {
  134.                     $definitions[$definitionId '.messenger.handler_descriptor.'.ContainerBuilder::hash($bus.':'.$message.':'.$handler[0])] = (new Definition(HandlerDescriptor::class))->setArguments([new Reference($handler[0]), $handler[1]]);
  135.                     $handlerDescriptors[] = new Reference($definitionId);
  136.                 }
  137.                 $handlersLocatorMappingByBus[$bus][$message] = new IteratorArgument($handlerDescriptors);
  138.             }
  139.         }
  140.         $container->addDefinitions($definitions);
  141.         foreach ($busIds as $bus) {
  142.             $container->register($locatorId $bus.'.messenger.handlers_locator'HandlersLocator::class)
  143.                 ->setArgument(0$handlersLocatorMappingByBus[$bus] ?? [])
  144.             ;
  145.             if ($container->has($handleMessageId $bus.'.middleware.handle_message')) {
  146.                 $container->getDefinition($handleMessageId)
  147.                     ->replaceArgument(0, new Reference($locatorId))
  148.                 ;
  149.             }
  150.         }
  151.         if ($container->hasDefinition('console.command.messenger_debug')) {
  152.             $debugCommandMapping $handlersByBusAndMessage;
  153.             foreach ($busIds as $bus) {
  154.                 if (!isset($debugCommandMapping[$bus])) {
  155.                     $debugCommandMapping[$bus] = [];
  156.                 }
  157.                 foreach ($debugCommandMapping[$bus] as $message => $handlers) {
  158.                     foreach ($handlers as $key => $handler) {
  159.                         $debugCommandMapping[$bus][$message][$key][0] = $handlerToOriginalServiceIdMapping[$handler[0]];
  160.                     }
  161.                 }
  162.             }
  163.             $container->getDefinition('console.command.messenger_debug')->replaceArgument(0$debugCommandMapping);
  164.         }
  165.     }
  166.     private function guessHandledClasses(\ReflectionClass $handlerClassstring $serviceIdstring $methodName): iterable
  167.     {
  168.         if ($handlerClass->implementsInterface(MessageSubscriberInterface::class)) {
  169.             trigger_deprecation('symfony/messenger''6.2''Implementing "%s" is deprecated, use the "%s" attribute instead.'MessageSubscriberInterface::class, AsMessageHandler::class);
  170.             return $handlerClass->getName()::getHandledMessages();
  171.         }
  172.         if ($handlerClass->implementsInterface(MessageHandlerInterface::class)) {
  173.             trigger_deprecation('symfony/messenger''6.2''Implementing "%s" is deprecated, use the "%s" attribute instead.'MessageHandlerInterface::class, AsMessageHandler::class);
  174.         }
  175.         try {
  176.             $method $handlerClass->getMethod($methodName);
  177.         } catch (\ReflectionException) {
  178.             throw new RuntimeException(sprintf('Invalid handler service "%s": class "%s" must have an "%s()" method.'$serviceId$handlerClass->getName(), $methodName));
  179.         }
  180.         if (=== $method->getNumberOfRequiredParameters()) {
  181.             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));
  182.         }
  183.         $parameters $method->getParameters();
  184.         /** @var \ReflectionNamedType|\ReflectionUnionType|null */
  185.         $type $parameters[0]->getType();
  186.         if (!$type) {
  187.             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));
  188.         }
  189.         if ($type instanceof \ReflectionUnionType) {
  190.             $types = [];
  191.             $invalidTypes = [];
  192.             foreach ($type->getTypes() as $type) {
  193.                 if (!$type->isBuiltin()) {
  194.                     $types[] = (string) $type;
  195.                 } else {
  196.                     $invalidTypes[] = (string) $type;
  197.                 }
  198.             }
  199.             if ($types) {
  200.                 return ('__invoke' === $methodName) ? $types array_fill_keys($types$methodName);
  201.             }
  202.             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)));
  203.         }
  204.         if ($type->isBuiltin()) {
  205.             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));
  206.         }
  207.         return ('__invoke' === $methodName) ? [$type->getName()] : [$type->getName() => $methodName];
  208.     }
  209.     private function registerReceivers(ContainerBuilder $container, array $busIds)
  210.     {
  211.         $receiverMapping = [];
  212.         $failureTransportsMap = [];
  213.         if ($container->hasDefinition('console.command.messenger_failed_messages_retry')) {
  214.             $commandDefinition $container->getDefinition('console.command.messenger_failed_messages_retry');
  215.             $globalReceiverName $commandDefinition->getArgument(0);
  216.             if (null !== $globalReceiverName) {
  217.                 if ($container->hasAlias('messenger.failure_transports.default')) {
  218.                     $failureTransportsMap[$globalReceiverName] = new Reference('messenger.failure_transports.default');
  219.                 } else {
  220.                     $failureTransportsMap[$globalReceiverName] = new Reference('messenger.transport.'.$globalReceiverName);
  221.                 }
  222.             }
  223.         }
  224.         foreach ($container->findTaggedServiceIds('messenger.receiver') as $id => $tags) {
  225.             $receiverClass $this->getServiceClass($container$id);
  226.             if (!is_subclass_of($receiverClassReceiverInterface::class)) {
  227.                 throw new RuntimeException(sprintf('Invalid receiver "%s": class "%s" must implement interface "%s".'$id$receiverClassReceiverInterface::class));
  228.             }
  229.             $receiverMapping[$id] = new Reference($id);
  230.             foreach ($tags as $tag) {
  231.                 if (isset($tag['alias'])) {
  232.                     $receiverMapping[$tag['alias']] = $receiverMapping[$id];
  233.                     if ($tag['is_failure_transport'] ?? false) {
  234.                         $failureTransportsMap[$tag['alias']] = $receiverMapping[$id];
  235.                     }
  236.                 }
  237.             }
  238.         }
  239.         $receiverNames = [];
  240.         foreach ($receiverMapping as $name => $reference) {
  241.             $receiverNames[(string) $reference] = $name;
  242.         }
  243.         $buses = [];
  244.         foreach ($busIds as $busId) {
  245.             $buses[$busId] = new Reference($busId);
  246.         }
  247.         if ($hasRoutableMessageBus $container->hasDefinition('messenger.routable_message_bus')) {
  248.             $container->getDefinition('messenger.routable_message_bus')
  249.                 ->replaceArgument(0ServiceLocatorTagPass::register($container$buses));
  250.         }
  251.         if ($container->hasDefinition('console.command.messenger_consume_messages')) {
  252.             $consumeCommandDefinition $container->getDefinition('console.command.messenger_consume_messages');
  253.             if ($hasRoutableMessageBus) {
  254.                 $consumeCommandDefinition->replaceArgument(0, new Reference('messenger.routable_message_bus'));
  255.             }
  256.             $consumeCommandDefinition->replaceArgument(4array_values($receiverNames));
  257.             try {
  258.                 $consumeCommandDefinition->replaceArgument(6$busIds);
  259.             } catch (OutOfBoundsException) {
  260.                 // ignore to preserve compatibility with symfony/framework-bundle < 5.4
  261.             }
  262.         }
  263.         if ($container->hasDefinition('console.command.messenger_setup_transports')) {
  264.             $container->getDefinition('console.command.messenger_setup_transports')
  265.                 ->replaceArgument(1array_values($receiverNames));
  266.         }
  267.         if ($container->hasDefinition('console.command.messenger_stats')) {
  268.             $container->getDefinition('console.command.messenger_stats')
  269.                 ->replaceArgument(1array_values($receiverNames));
  270.         }
  271.         $container->getDefinition('messenger.receiver_locator')->replaceArgument(0$receiverMapping);
  272.         $failureTransportsLocator ServiceLocatorTagPass::register($container$failureTransportsMap);
  273.         $failedCommandIds = [
  274.             'console.command.messenger_failed_messages_retry',
  275.             'console.command.messenger_failed_messages_show',
  276.             'console.command.messenger_failed_messages_remove',
  277.         ];
  278.         foreach ($failedCommandIds as $failedCommandId) {
  279.             if ($container->hasDefinition($failedCommandId)) {
  280.                 $definition $container->getDefinition($failedCommandId);
  281.                 $definition->replaceArgument(1$failureTransportsLocator);
  282.             }
  283.         }
  284.     }
  285.     private function registerBusToCollector(ContainerBuilder $containerstring $busId)
  286.     {
  287.         $container->setDefinition(
  288.             $tracedBusId 'debug.traced.'.$busId,
  289.             (new Definition(TraceableMessageBus::class, [new Reference($tracedBusId.'.inner')]))->setDecoratedService($busId)
  290.         );
  291.         $container->getDefinition('data_collector.messenger')->addMethodCall('registerBus', [$busId, new Reference($tracedBusId)]);
  292.     }
  293.     private function registerBusMiddleware(ContainerBuilder $containerstring $busId, array $middlewareCollection)
  294.     {
  295.         $middlewareReferences = [];
  296.         foreach ($middlewareCollection as $middlewareItem) {
  297.             $id $middlewareItem['id'];
  298.             $arguments $middlewareItem['arguments'] ?? [];
  299.             if (!$container->has($messengerMiddlewareId 'messenger.middleware.'.$id)) {
  300.                 $messengerMiddlewareId $id;
  301.             }
  302.             if (!$container->has($messengerMiddlewareId)) {
  303.                 throw new RuntimeException(sprintf('Invalid middleware: service "%s" not found.'$id));
  304.             }
  305.             if ($container->findDefinition($messengerMiddlewareId)->isAbstract()) {
  306.                 $childDefinition = new ChildDefinition($messengerMiddlewareId);
  307.                 $childDefinition->setArguments($arguments);
  308.                 if (isset($middlewareReferences[$messengerMiddlewareId $busId.'.middleware.'.$id])) {
  309.                     $messengerMiddlewareId .= '.'.ContainerBuilder::hash($arguments);
  310.                 }
  311.                 $container->setDefinition($messengerMiddlewareId$childDefinition);
  312.             } elseif ($arguments) {
  313.                 throw new RuntimeException(sprintf('Invalid middleware factory "%s": a middleware factory must be an abstract definition.'$id));
  314.             }
  315.             $middlewareReferences[$messengerMiddlewareId] = new Reference($messengerMiddlewareId);
  316.         }
  317.         $container->getDefinition($busId)->replaceArgument(0, new IteratorArgument(array_values($middlewareReferences)));
  318.     }
  319.     private function getServiceClass(ContainerBuilder $containerstring $serviceId): string
  320.     {
  321.         while (true) {
  322.             $definition $container->findDefinition($serviceId);
  323.             if (!$definition->getClass() && $definition instanceof ChildDefinition) {
  324.                 $serviceId $definition->getParent();
  325.                 continue;
  326.             }
  327.             return $definition->getClass();
  328.         }
  329.     }
  330. }