MessengerPass.php 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402
  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\Handler\HandlerDescriptor;
  21. use Symfony\Component\Messenger\Handler\HandlersLocator;
  22. use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
  23. use Symfony\Component\Messenger\TraceableMessageBus;
  24. use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
  25. /**
  26. * @author Samuel Roze <samuel.roze@gmail.com>
  27. */
  28. class MessengerPass implements CompilerPassInterface
  29. {
  30. private $handlerTag;
  31. private $busTag;
  32. private $receiverTag;
  33. public function __construct(string $handlerTag = 'messenger.message_handler', string $busTag = 'messenger.bus', string $receiverTag = 'messenger.receiver')
  34. {
  35. if (0 < \func_num_args()) {
  36. trigger_deprecation('symfony/messenger', '5.3', 'Configuring "%s" is deprecated.', __CLASS__);
  37. }
  38. $this->handlerTag = $handlerTag;
  39. $this->busTag = $busTag;
  40. $this->receiverTag = $receiverTag;
  41. }
  42. /**
  43. * {@inheritdoc}
  44. */
  45. public function process(ContainerBuilder $container)
  46. {
  47. $busIds = [];
  48. foreach ($container->findTaggedServiceIds($this->busTag) as $busId => $tags) {
  49. $busIds[] = $busId;
  50. if ($container->hasParameter($busMiddlewareParameter = $busId.'.middleware')) {
  51. $this->registerBusMiddleware($container, $busId, $container->getParameter($busMiddlewareParameter));
  52. $container->getParameterBag()->remove($busMiddlewareParameter);
  53. }
  54. if ($container->hasDefinition('data_collector.messenger')) {
  55. $this->registerBusToCollector($container, $busId);
  56. }
  57. }
  58. if ($container->hasDefinition('messenger.receiver_locator')) {
  59. $this->registerReceivers($container, $busIds);
  60. }
  61. $this->registerHandlers($container, $busIds);
  62. }
  63. private function registerHandlers(ContainerBuilder $container, array $busIds)
  64. {
  65. $definitions = [];
  66. $handlersByBusAndMessage = [];
  67. $handlerToOriginalServiceIdMapping = [];
  68. foreach ($container->findTaggedServiceIds($this->handlerTag, true) as $serviceId => $tags) {
  69. foreach ($tags as $tag) {
  70. if (isset($tag['bus']) && !\in_array($tag['bus'], $busIds, true)) {
  71. throw new RuntimeException(sprintf('Invalid handler service "%s": bus "%s" specified on the tag "%s" does not exist (known ones are: "%s").', $serviceId, $tag['bus'], $this->handlerTag, implode('", "', $busIds)));
  72. }
  73. $className = $this->getServiceClass($container, $serviceId);
  74. $r = $container->getReflectionClass($className);
  75. if (null === $r) {
  76. throw new RuntimeException(sprintf('Invalid service "%s": class "%s" does not exist.', $serviceId, $className));
  77. }
  78. if (isset($tag['handles'])) {
  79. $handles = isset($tag['method']) ? [$tag['handles'] => $tag['method']] : [$tag['handles']];
  80. } else {
  81. $handles = $this->guessHandledClasses($r, $serviceId);
  82. }
  83. $message = null;
  84. $handlerBuses = (array) ($tag['bus'] ?? $busIds);
  85. foreach ($handles as $message => $options) {
  86. $buses = $handlerBuses;
  87. if (\is_int($message)) {
  88. if (\is_string($options)) {
  89. $message = $options;
  90. $options = [];
  91. } else {
  92. 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));
  93. }
  94. }
  95. if (\is_string($options)) {
  96. $options = ['method' => $options];
  97. }
  98. if (!isset($options['from_transport']) && isset($tag['from_transport'])) {
  99. $options['from_transport'] = $tag['from_transport'];
  100. }
  101. $priority = $tag['priority'] ?? $options['priority'] ?? 0;
  102. $method = $options['method'] ?? '__invoke';
  103. if (isset($options['bus'])) {
  104. if (!\in_array($options['bus'], $busIds)) {
  105. $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));
  106. throw new RuntimeException(sprintf('Invalid configuration "%s" for message "%s": bus "%s" does not exist.', $messageLocation, $message, $options['bus']));
  107. }
  108. $buses = [$options['bus']];
  109. }
  110. if ('*' !== $message && !class_exists($message) && !interface_exists($message, false)) {
  111. $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));
  112. throw new RuntimeException(sprintf('Invalid handler service "%s": class or interface "%s" "%s" not found.', $serviceId, $message, $messageLocation));
  113. }
  114. if (!$r->hasMethod($method)) {
  115. throw new RuntimeException(sprintf('Invalid handler service "%s": method "%s::%s()" does not exist.', $serviceId, $r->getName(), $method));
  116. }
  117. if ('__invoke' !== $method) {
  118. $wrapperDefinition = (new Definition('callable'))->addArgument([new Reference($serviceId), $method])->setFactory('Closure::fromCallable');
  119. $definitions[$definitionId = '.messenger.method_on_object_wrapper.'.ContainerBuilder::hash($message.':'.$priority.':'.$serviceId.':'.$method)] = $wrapperDefinition;
  120. } else {
  121. $definitionId = $serviceId;
  122. }
  123. $handlerToOriginalServiceIdMapping[$definitionId] = $serviceId;
  124. foreach ($buses as $handlerBus) {
  125. $handlersByBusAndMessage[$handlerBus][$message][$priority][] = [$definitionId, $options];
  126. }
  127. }
  128. if (null === $message) {
  129. throw new RuntimeException(sprintf('Invalid handler service "%s": method "%s::getHandledMessages()" must return one or more messages.', $serviceId, $r->getName()));
  130. }
  131. }
  132. }
  133. foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
  134. foreach ($handlersByMessage as $message => $handlersByPriority) {
  135. krsort($handlersByPriority);
  136. $handlersByBusAndMessage[$bus][$message] = array_merge(...$handlersByPriority);
  137. }
  138. }
  139. $handlersLocatorMappingByBus = [];
  140. foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
  141. foreach ($handlersByMessage as $message => $handlers) {
  142. $handlerDescriptors = [];
  143. foreach ($handlers as $handler) {
  144. $definitions[$definitionId = '.messenger.handler_descriptor.'.ContainerBuilder::hash($bus.':'.$message.':'.$handler[0])] = (new Definition(HandlerDescriptor::class))->setArguments([new Reference($handler[0]), $handler[1]]);
  145. $handlerDescriptors[] = new Reference($definitionId);
  146. }
  147. $handlersLocatorMappingByBus[$bus][$message] = new IteratorArgument($handlerDescriptors);
  148. }
  149. }
  150. $container->addDefinitions($definitions);
  151. foreach ($busIds as $bus) {
  152. $container->register($locatorId = $bus.'.messenger.handlers_locator', HandlersLocator::class)
  153. ->setArgument(0, $handlersLocatorMappingByBus[$bus] ?? [])
  154. ;
  155. if ($container->has($handleMessageId = $bus.'.middleware.handle_message')) {
  156. $container->getDefinition($handleMessageId)
  157. ->replaceArgument(0, new Reference($locatorId))
  158. ;
  159. }
  160. }
  161. if ($container->hasDefinition('console.command.messenger_debug')) {
  162. $debugCommandMapping = $handlersByBusAndMessage;
  163. foreach ($busIds as $bus) {
  164. if (!isset($debugCommandMapping[$bus])) {
  165. $debugCommandMapping[$bus] = [];
  166. }
  167. foreach ($debugCommandMapping[$bus] as $message => $handlers) {
  168. foreach ($handlers as $key => $handler) {
  169. $debugCommandMapping[$bus][$message][$key][0] = $handlerToOriginalServiceIdMapping[$handler[0]];
  170. }
  171. }
  172. }
  173. $container->getDefinition('console.command.messenger_debug')->replaceArgument(0, $debugCommandMapping);
  174. }
  175. }
  176. private function guessHandledClasses(\ReflectionClass $handlerClass, string $serviceId): iterable
  177. {
  178. if ($handlerClass->implementsInterface(MessageSubscriberInterface::class)) {
  179. return $handlerClass->getName()::getHandledMessages();
  180. }
  181. try {
  182. $method = $handlerClass->getMethod('__invoke');
  183. } catch (\ReflectionException $e) {
  184. throw new RuntimeException(sprintf('Invalid handler service "%s": class "%s" must have an "__invoke()" method.', $serviceId, $handlerClass->getName()));
  185. }
  186. if (0 === $method->getNumberOfRequiredParameters()) {
  187. throw new RuntimeException(sprintf('Invalid handler service "%s": method "%s::__invoke()" requires at least one argument, first one being the message it handles.', $serviceId, $handlerClass->getName()));
  188. }
  189. $parameters = $method->getParameters();
  190. if (!$type = $parameters[0]->getType()) {
  191. throw new RuntimeException(sprintf('Invalid handler service "%s": argument "$%s" of method "%s::__invoke()" must have a type-hint corresponding to the message class it handles.', $serviceId, $parameters[0]->getName(), $handlerClass->getName()));
  192. }
  193. if ($type instanceof \ReflectionUnionType) {
  194. $types = [];
  195. $invalidTypes = [];
  196. foreach ($type->getTypes() as $type) {
  197. if (!$type->isBuiltin()) {
  198. $types[] = (string) $type;
  199. } else {
  200. $invalidTypes[] = (string) $type;
  201. }
  202. }
  203. if ($types) {
  204. return $types;
  205. }
  206. 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)));
  207. }
  208. if ($type->isBuiltin()) {
  209. 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(), $type instanceof \ReflectionNamedType ? $type->getName() : (string) $type));
  210. }
  211. return [$type->getName()];
  212. }
  213. private function registerReceivers(ContainerBuilder $container, array $busIds)
  214. {
  215. $receiverMapping = [];
  216. $failureTransportsMap = [];
  217. if ($container->hasDefinition('console.command.messenger_failed_messages_retry')) {
  218. $commandDefinition = $container->getDefinition('console.command.messenger_failed_messages_retry');
  219. $globalReceiverName = $commandDefinition->getArgument(0);
  220. if (null !== $globalReceiverName) {
  221. if ($container->hasAlias('messenger.failure_transports.default')) {
  222. $failureTransportsMap[$globalReceiverName] = new Reference('messenger.failure_transports.default');
  223. } else {
  224. $failureTransportsMap[$globalReceiverName] = new Reference('messenger.transport.'.$globalReceiverName);
  225. }
  226. }
  227. }
  228. foreach ($container->findTaggedServiceIds($this->receiverTag) as $id => $tags) {
  229. $receiverClass = $this->getServiceClass($container, $id);
  230. if (!is_subclass_of($receiverClass, ReceiverInterface::class)) {
  231. throw new RuntimeException(sprintf('Invalid receiver "%s": class "%s" must implement interface "%s".', $id, $receiverClass, ReceiverInterface::class));
  232. }
  233. $receiverMapping[$id] = new Reference($id);
  234. foreach ($tags as $tag) {
  235. if (isset($tag['alias'])) {
  236. $receiverMapping[$tag['alias']] = $receiverMapping[$id];
  237. if ($tag['is_failure_transport'] ?? false) {
  238. $failureTransportsMap[$tag['alias']] = $receiverMapping[$id];
  239. }
  240. }
  241. }
  242. }
  243. $receiverNames = [];
  244. foreach ($receiverMapping as $name => $reference) {
  245. $receiverNames[(string) $reference] = $name;
  246. }
  247. $buses = [];
  248. foreach ($busIds as $busId) {
  249. $buses[$busId] = new Reference($busId);
  250. }
  251. if ($hasRoutableMessageBus = $container->hasDefinition('messenger.routable_message_bus')) {
  252. $container->getDefinition('messenger.routable_message_bus')
  253. ->replaceArgument(0, ServiceLocatorTagPass::register($container, $buses));
  254. }
  255. if ($container->hasDefinition('console.command.messenger_consume_messages')) {
  256. $consumeCommandDefinition = $container->getDefinition('console.command.messenger_consume_messages');
  257. if ($hasRoutableMessageBus) {
  258. $consumeCommandDefinition->replaceArgument(0, new Reference('messenger.routable_message_bus'));
  259. }
  260. $consumeCommandDefinition->replaceArgument(4, array_values($receiverNames));
  261. try {
  262. $consumeCommandDefinition->replaceArgument(6, $busIds);
  263. } catch (OutOfBoundsException $e) {
  264. // ignore to preserve compatibility with symfony/framework-bundle < 5.4
  265. }
  266. }
  267. if ($container->hasDefinition('console.command.messenger_setup_transports')) {
  268. $container->getDefinition('console.command.messenger_setup_transports')
  269. ->replaceArgument(1, array_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 $container, string $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 $container, string $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 $container, string $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. }