8000 Allows to register handlers on a specific transport (and get rid of t… · symfony/symfony@106be1d · GitHub
[go: up one dir, main page]
Skip to content

Commit 106be1d

Browse files
srozesroze
committed
Allows to register handlers on a specific transport (and get rid of this handler alias)
1 parent f82f1c0 commit 106be1d

22 files changed

+417
-210
lines changed

‎src/Symfony/Component/Messenger/Command/DebugCommand.php

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
use Symfony\Component\Console\Input\InputInterface;
1818
use Symfony\Component\Console\Output\OutputInterface;
1919
use Symfony\Component\Console\Style\SymfonyStyle;
20+
use Symfony\Component\Messenger\Handler\HandlerDescriptor;
2021

2122
/**
2223
* A console command to debug Messenger information.
@@ -84,7 +85,9 @@ protected function execute(InputInterface $input, OutputInterface $output)
8485
foreach ($handlersByMessage as $message => $handlers) {
8586
$tableRows[] = [sprintf('<fg=cyan>%s</fg=cyan>', $message)];
8687
foreach ($handlers as $handler) {
87-
$tableRows[] = [sprintf(' handled by <info>%s</>', $handler)];
88+
$tableRows[] = [
89+
sprintf(' handled by <info>%s</>', $handler[0]).$this->formatConditions($handler[1]),
90+
];
8891
}
8992
}
9093

@@ -97,4 +100,18 @@ protected function execute(InputInterface $input, OutputInterface $output)
97100
}
98101
}
99102
}
103+
104+
private function formatConditions(array $options): string
105+
{
106+
if (empty($options)) {
107+
return '';
108+
}
109+
110+
$optionsMapping = [];
111+
foreach ($options as $key => $value) {
112+
$optionsMapping[] = ' '.$key.'='.$value;
113+
}
114+
115+
return ' (when'.implode(', ', $optionsMapping).')';
116+
}
100117
}

‎src/Symfony/Component/Messenger/DependencyInjection/MessengerPass.php

Lines changed: 30 additions & 21 deletions
use Symfony\Component\DependencyInjection\Exception\RuntimeException;
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
use Symfony\Component\DependencyInjection\Definition;
2020
2121
use Symfony\Component\DependencyInjection\Reference;
22+
use Symfony\Component\Messenger\Handler\HandlerDescriptor;
2223
use Symfony\Component\Messenger\Handler\HandlersLocator;
2324
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
2425
use Symfony\Component\Messenger\TraceableMessageBus;
@@ -94,34 +95,36 @@ private function registerHandlers(ContainerBuilder $container, array $busIds)
9495
$message = null;
9596
$handlerBuses = (array) ($tag['bus'] ?? $busIds);
9697

97-
foreach ($handles as $message => $method) {
98+
foreach ($handles as $message => $options) {
9899
$buses = $handlerBuses;
100+
99101
if (\is_int($message)) {
100-
$message = $method;
101-
$method = '__invoke';
102+
if (\is_string($options)) {
103+
$message = $options;
104+
$options = [];
105+
} else {
106+
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".', \gettype($options), $message, $serviceId));
107+
}
102108
}
103109

104-
if (\is_array($message)) {
105-
list($message, $priority) = $message;
106-
} else {
107-
$priority = $tag['priority'] ?? 0;
110+
if (\is_string($options)) {
111+
$options = ['method' => $options];
108112
}
109113

110-
if (\is_array($method)) {
111-
if (isset($method['bus'])) {
112-
if (!\in_array($method['bus'], $busIds)) {
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);
114+
$priority = $tag['priority'] ?? $options['priority'] ?? 0;
115+
$method = $options['method'] ?? '__invoke';
114116

115-
throw new RuntimeException(sprintf('Invalid configuration %s for message "%s": bus "%s" does not exist.', $messageLocation, $message, $method['bus']));
116-
}
117+
if (isset($options['bus'])) {
118+
if (!\in_array($options['bus'], $busIds)) {
119+
$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);
117120

118-
$buses = [$method['bus']];
121+
throw new RuntimeException(sprintf('Invalid configuration %s for message "%s": bus "%s" does not exist.', $messageLocation, $message, $options['bus']));
119122
}
120123

121-
$priority = $method['priority'] ?? $priority;
122-
$method = $method['method'] ?? '__invoke';
124+
$buses = [$options['bus']];
123125
}
124126

127+
125128
if ('*' !== $message && !class_exists($message) && !interface_exists($message, false)) {
126129
$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);
127130

@@ -141,7 +144,7 @@ private function registerHandlers(ContainerBuilder $container, array $busIds)
141144
}
142145

143146
foreach ($buses as $handlerBus) {
144-
$handlersByBusAndMessage[$handlerBus][$message][$priority][] = $definitionId;
147+
$handlersByBusAndMessage[$handlerBus][$message][$priority][] = [$definitionId, $options];
145148
}
146149
}
147150

@@ -154,15 +157,21 @@ private function registerHandlers(ContainerBuilder $container, array $busIds)
154157
foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
155158
foreach ($handlersByMessage as $message => $handlersByPriority) {
156159
krsort($handlersByPriority);
157-
$handlersByBusAndMessage[$bus][$message] = array_unique(array_merge(...$handlersByPriority));
160+
$handlersByBusAndMessage[$bus][$message] = array_merge(...$handlersByPriority);
158161
}
159162
}
160163

161164
$handlersLocatorMappingByBus = [];
162165
foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
163-
foreach ($handlersByMessage as $message => $handlerIds) {
164-
$handlers = array_map(function (string $handlerId) { return new Reference($handlerId); }, $handlerIds);
165-
$handlersLocatorMappingByBus[$bus][$message] = new IteratorArgument($handlers);
166+
foreach ($handlersByMessage as $message => $handlers) {
167+
168+
$handlerDescriptors = [];
169+
foreach ($handlers as $handler) {
170+
$definitions[$definitionId = '.messenger.handler_descriptor.'.ContainerBuilder::hash($bus.':'.$message.':'.$handler[0])] = (new Definition(HandlerDescriptor::class))->setArguments([new Reference($handler[0]), $handler[1]])->setFactory([HandlerDescriptor::class, 'create']);
171+
$handlerDescriptors[] = new Reference($definitionId);
172+
}
173+
174+
$handlersLocatorMappingByBus[$bus][$message] = new IteratorArgument($handlerDescriptors);
166175
}
167176
}
168177
$container->addDefinitions($definitions);

‎src/Symfony/Component/Messenger/HandleTrait.php

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ private function handle($message)
5252

5353
if (\count($handledStamps) > 1) {
5454
$handlers = implode(', ', array_map(function (HandledStamp $stamp): string {
55-
return sprintf('"%s"', $stamp->getHandlerAlias() ?? $stamp->getCallableName());
55+
return sprintf('"%s"', $stamp->getHandlerName());
5656
}, $handledStamps));
5757

5858
throw new LogicException(sprintf('Message of type "%s" was handled multiple times. Only one handler is expected when using "%s::%s()", got %d: %s.', \get_class($envelope->getMessage()), \get_class($this), __FUNCTION__, \count($handledStamps), $handlers));

‎src/Symfony/Component/Messenger/Handler/HandlerDescriptor.php

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
<?php
2+
3+
namespace Symfony\Component\Messenger\Handler;
4+
5+
/**
6+
*
7+
* @author Samuel Rozeé
8+
*
9+
* @final
10+
*/
11+
class HandlerDescriptor
12+
{
13+
private $handler;
14+
private $options;
15+
16+
public function __construct(callable $handler, array $options = [])
17+
{
18+
$this->handler = $handler;
19+
$this->options = $options;
20+
}
21+
22+
public function getHandler(): callable
23+
{
24+
return $this->handler;
25+
}
26+
27+
public function getName(): string
28+
{
29+
$name = $this->callableName($this->handler);
30+
$alias = $this->options['alias'] ?? null;
31+
32+
if (null !== $alias) {
33+
$name .= '@'.$alias;
34+
}
35+
36+
return $name;
37+
}
38+
39+
private function callableName(callable $handler)
40+
{
41+
if (\is_array($handler)) {
42+
if (\is_object($handler[0])) {
43+
return \get_class($handler[0]).'::'.$handler[1];
44+
}
45+
46+
return $handler[0].'::'.$handler[1];
47+
}
48+
49+
if (\is_string($handler)) {
50+
return $handler;
51+
}
52+
53+
if ($handler instanceof \Closure) {
54+
$r = new \ReflectionFunction($handler);
55+
if (false !== strpos($r->name, '{closure}')) {
56+
return 'Closure';
57+
}
58+
if ($class = $r->getClosureScopeClass()) {
59+
return $class->name.'::'.$r->name;
60+
}
61+
62+
return $r->name;
63+
}
64+
65+
return \get_class($handler).'::__invoke';
66+
}
67+
68+
public function getOption(string $option)
69+
{
70+
return $this->options[$option] ?? null;
71+
}
72+
}

‎src/Symfony/Component/Messenger/Handler/HandlersLocator.php

Lines changed: 32 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -12,11 +12,13 @@
1212
namespace Symfony\Component\Messenger\Handler;
1313

1414
use Symfony\Component\Messenger\Envelope;
15+
use Symfony\Component\Messenger\Stamp\ReceivedStamp;
1516

1617
/**
1718
* Maps a message to a list of handlers.
1819
*
1920
* @author Nicolas Grekas <p@tchwork.com>
21+
* @author Samuel Roze <samuel.roze@gmail.com>
2022
*
2123
* @experimental in 4.2
2224
*/
@@ -25,7 +27,7 @@ class HandlersLocator implements HandlersLocatorInterface
2527
private $handlers;
2628

2729
/**
28-
* @param callable[][] $handlers
30+
* @param HandlerDescriptor[][]|callable[][] $handlers
2931
*/
3032
public function __construct(array $handlers)
3133
{
@@ -40,10 +42,23 @@ public function getHandlers(Envelope $envelope): iterable
4042
$seen = [];
4143

4244
foreach (self::listTypes($envelope) as $type) {
43-
foreach ($this->handlers[$type] ?? [] as $alias => $handler) {
44-
if (!\in_array($handler, $seen, true)) {
45-
yield $alias => $seen[] = $handler;
45+
foreach ($this->handlers[$type] ?? [] as $handlerDescriptor) {
46+
if (is_callable($handlerDescriptor)) {
47+
$handlerDescriptor = new HandlerDescriptor($handlerDescriptor);
4648
}
49+
50+
if (!$this->shouldHandle($envelope, $handlerDescriptor)) {
51+
continue;
52+
}
53+
54+
$name = $handlerDescriptor->getName();
55+
if (\in_array($name, $seen)) {
56+
continue;
57+
}
58+
59+
$seen[] = $name;
60+
61+
yield $handlerDescriptor;
4762
}
4863
}
4964
}
@@ -60,4 +75,17 @@ public static function listTypes(Envelope $envelope): array
6075
+ class_implements($class)
6176
+ ['*' => '*'];
6277
}
78+
79+
private function shouldHandle(Envelope $envelope, HandlerDescriptor $handlerDescriptor)
80+
{
81+
if (null === $received = $envelope->last(ReceivedStamp::class)) {
82+
return true;
83+
}
84+
85+
if (null === $expectedTransport = $handlerDescriptor->getOption('transport')) {
86+
return true;
87+
}
88+
89+
return $received->getTransportName() === $expectedTransport;
90+
}
6391
}

< 427E a class="Link--primary prc-Link-Link-85e08" href="#diff-d13297093e29942dcc737e293418524f1f9f3abd369ba5d7e6f20a5e34c95b82" data-analytics-opt-out="true">‎src/Symfony/Component/Messenger/Handler/HandlersLocatorInterface.php

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ interface HandlersLocatorInterface
2525
/**
2626
* Returns the handlers for the given message name.
2727
*
28-
* @return iterable|callable[] Indexed by handler alias if available
28+
* @return iterable|HandlerDescriptor[] Indexed by handler alias if available
2929
*/
3030
public function getHandlers(Envelope $envelope): iterable;
3131
}

‎src/Symfony/Component/Messenger/Middleware/HandleMessageMiddleware.php

Lines changed: 9 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
use Symfony\Component\Messenger\Envelope;
1717
use Symfony\Component\Messenger\Exception\HandlerFailedException;
1818
use Symfony\Component\Messenger\Exception\NoHandlerForMessageException;
19+
use Symfony\Component\Messenger\Handler\HandlerDescriptor;
1920
use Symfony\Component\Messenger\Handler\HandlersLocatorInterface;
2021
use Symfony\Component\Messenger\Stamp\HandledStamp;
2122

@@ -54,17 +55,16 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
5455
];
5556

5657
$exceptions = [];
57-
foreach ($this->handlersLocator->getHandlers($envelope) as $alias => $handler) {
58-
$alias = \is_string($alias) ? $alias : null;
59-
60-
if ($this->messageHasAlreadyBeenHandled($envelope, $handler, $alias)) {
58+
foreach ($this->handlersLocator->getHandlers($envelope) as $handlerDescriptor) {
59+
if ($this->messageHasAlreadyBeenHandled($envelope, $handlerDescriptor)) {
6160
continue;
6261
}
6362

6463
try {
65-
$handledStamp = HandledStamp::fromCallable($handler, $handler($message), $alias);
64+
$handler = $handlerDescriptor->getHandler();
65+
$handledStamp = HandledStamp::fromDescriptor($handlerDescriptor, $handler($message));
6666
$envelope = $envelope->with($handledStamp);
67-
$this->logger->info('Message "{class}" handled by "{handler}"', $context + ['handler' => $handledStamp->getCallableName()]);
67+
$this->logger->info('Message "{class}" handled by "{handler}"', $context + ['handler' => $handledStamp->getHandlerName()]);
6868
} catch (\Throwable $e) {
6969
$exceptions[] = $e;
7070
}
@@ -85,12 +85,11 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
8585
return $stack->next()->handle($envelope, $stack);
8686
}
8787

88-
private function messageHasAlreadyBeenHandled(Envelope $envelope, callable $handler, ?string $alias): bool
88+
private function messageHasAlreadyBeenHandled(Envelope $envelope, HandlerDescriptor $handlerDescriptor): bool
8989
{
9090
$some = array_filter($envelope
91-
->all(HandledStamp::class), function (HandledStamp $stamp) use ($handler, $alias) {
92-
return $stamp->getCallableName() === HandledStamp::getNameFromCallable($handler) &&
93-
$stamp->getHandlerAlias() === $alias;
91+
->all(HandledStamp::class), function (HandledStamp $stamp) use ($handlerDescriptor) {
92+
return $stamp->getHandlerName() === $handlerDescriptor->getName();
9493
});
9594

9695
return \count($some) > 0;

0 commit comments

Comments
 (0)
0