Skip to content

Commit f0b3421

Browse files
srozesroze
committed
feature #30958 [Messenger] Allows to register handlers on a specific transport (sroze)
This PR was merged into the 4.3-dev branch. Discussion ---------- [Messenger] Allows to register handlers on a specific transport | Q | A | ------------- | --- | Branch? | master | Bug fix? | no | New feature? | yes | BC breaks? | no | Deprecations? | no | Tests pass? | yes | Fixed tickets | #30110 | License | MIT | Doc PR | symfony/symfony-docs#11236 With the #30008 pull-request in, we can now do the following: ```yaml framework: messenger: transports: events: dsn: amqp://guest:guest@127.0.0.1/%2f/events options: queues: 3rdparty: binding_keys: [ 3rdparty ] projection: binding_keys: [ projection ] events_3rdparty: amqp://guest:guest@127.0.0.1/%2f/events?queues[3rdparty] events_projection: amqp://guest:guest@127.0.0.1/%2f/events?queues[projection] routing: 'App\Message\RegisterBet': events ``` This will bind two queues to the `events` exchange, fantastic: <img width="325" alt="Screenshot 2019-04-07 at 10 26 27" src="https://user-images.githubusercontent.com/804625/55680861-af373580-591f-11e9-8f1e-2d3b6ddba2fd.png"> --- Now, in this setup, the message will be duplicated within the `3rdparty` & `projection` queues. If you just run the consumer for each transport, it will consume the message and call all the handlers. You can't do different things based on which queue you have consumed the message. **This pull-request adds the following feature:** ```php class RegisterBetHandler implements MessageSubscriberInterface { public function __invoke(RegisterBet $message) { // Do something only when the message comes from the `events_projection` transport... } /** * {@inheritdoc} */ public static function getHandledMessages(): iterable { yield RegisterBet::class => [ 'from_transport' => 'events_projection', ]; } } ``` --- In the debugger, it looks like this: <img width="649" alt="Screenshot 2019-04-07 at 10 29 55" src="https://user-images.githubusercontent.com/804625/55680892-1d7bf800-5920-11e9-80f7-853f0201c6d8.png"> Commits ------- f0b2acd67d Allows to register handlers on a specific transport (and get rid of this handler alias)
2 parents 55661af + 0082f93 commit f0b3421

23 files changed

+426
-209
lines changed

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,12 @@ CHANGELOG
8181
* Added a Doctrine transport. For example, use the `doctrine://default` DSN (this uses the `default` Doctrine entity manager)
8282
* [BC BREAK] The `getConnectionConfiguration` method on Amqp's `Connection` has been removed.
8383
* [BC BREAK] A `HandlerFailedException` exception will be thrown if one or more handler fails.
84+
* [BC BREAK] The `HandlersLocationInterface::getHandlers` method needs to return `HandlerDescriptor`
85+
instances instead of callables.
86+
* [BC BREAK] The `HandledStamp` stamp has changed: `handlerAlias` has been renamed to `handlerName`,
87+
`getCallableName` has been removed and its constructor only has 2 arguments now.
88+
* [BC BREAK] The `ReceivedStamp` needs to exposes the name of the transport from which the message
89+
has been received.
8490

8591
4.2.0
8692
-----

‎Command/DebugCommand.php‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,9 @@ protected function execute(InputInterface $input, OutputInterface $output)
8484
foreach ($handlersByMessage as $message => $handlers) {
8585
$tableRows[] = [sprintf('<fg=cyan>%s</fg=cyan>', $message)];
8686
foreach ($handlers as $handler) {
87-
$tableRows[] = [sprintf(' handled by <info>%s</>', $handler)];
87+
$tableRows[] = [
88+
sprintf(' handled by <info>%s</>', $handler[0]).$this->formatConditions($handler[1]),
89+
];
8890
}
8991
}
9092

@@ -97,4 +99,18 @@ protected function execute(InputInterface $input, OutputInterface $output)
9799
}
98100
}
99101
}
102+
103+
private function formatConditions(array $options): string
104+
{
105+
if (!$options) {
106+
return '';
107+
}
108+
109+
$optionsMapping = [];
110+
foreach ($options as $key => $value) {
111+
$optionsMapping[] = ' '.$key.'='.$value;
112+
}
113+
114+
return ' (when'.implode(', ', $optionsMapping).')';
115+
}
100116
}

‎DependencyInjection/MessengerPass.php‎

Lines changed: 28 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
use Symfony\Component\DependencyInjection\Definition;
2020
use Symfony\Component\DependencyInjection\Exception\RuntimeException;
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,32 +95,33 @@ 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

125127
if ('*' !== $message && !class_exists($message) && !interface_exists($message, false)) {
@@ -141,7 +143,7 @@ private function registerHandlers(ContainerBuilder $container, array $busIds)
141143
}
142144

143145
foreach ($buses as $handlerBus) {
144-
$handlersByBusAndMessage[$handlerBus][$message][$priority][] = $definitionId;
146+
$handlersByBusAndMessage[$handlerBus][$message][$priority][] = [$definitionId, $options];
145147
}
146148
}
147149

@@ -154,15 +156,20 @@ private function registerHandlers(ContainerBuilder $container, array $busIds)
154156
foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
155157
foreach ($handlersByMessage as $message => $handlersByPriority) {
156158
krsort($handlersByPriority);
157-
$handlersByBusAndMessage[$bus][$message] = array_unique(array_merge(...$handlersByPriority));
159+
$handlersByBusAndMessage[$bus][$message] = array_merge(...$handlersByPriority);
158160
}
159161
}
160162

161163
$handlersLocatorMappingByBus = [];
162164
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);
165+
foreach ($handlersByMessage as $message => $handlers) {
166+
$handlerDescriptors = [];
167+
foreach ($handlers as $handler) {
168+
$definitions[$definitionId = '.messenger.handler_descriptor.'.ContainerBuilder::hash($bus.':'.$message.':'.$handler[0])] = (new Definition(HandlerDescriptor::class))->setArguments([new Reference($handler[0]), $handler[1]]);
169+
$handlerDescriptors[] = new Reference($definitionId);
170+
}
171+
172+
$handlersLocatorMappingByBus[$bus][$message] = new IteratorArgument($handlerDescriptors);
166173
}
167174
}
168175
$container->addDefinitions($definitions);

‎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));

‎Handler/HandlerDescriptor.php‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
<?php
2+
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+
*/
11+
12+
namespace Symfony\Component\Messenger\Handler;
13+
14+
/**
15+
* Describes a handler and the possible associated options, such as `from_transport`, `bus`, etc.
16+
*
17+
* @author Samuel Roze <samuel.roze@gmail.com>
18+
*
19+
* @experimental in 4.3
20+
*/
21+
final class HandlerDescriptor
22+
{
23+
private $handler;
24+
private $options;
25+
26+
public function __construct(callable $handler, array $options = [])
27+
{
28+
$this->handler = $handler;
29+
$this->options = $options;
30+
}
31+
32+
public function getHandler(): callable
33+
{
34+
return $this->handler;
35+
}
36+
37+
public function getName(): string
38+
{
39+
$name = $this->callableName($this->handler);
40+
$alias = $this->options['alias'] ?? null;
41+
42+
if (null !== $alias) {
43+
$name .= '@'.$alias;
44+
}
45+
46+
return $name;
47+
}
48+
49+
public function getOption(string $option)
50+
{
51+
return $this->options[$option] ?? null;
52+
}
53+
54+
private function callableName(callable $handler)
55+
{
56+
if (\is_array($handler)) {
57+
if (\is_object($handler[0])) {
58+
return \get_class($handler[0]).'::'.$handler[1];
59+
}
60+
61+
return $handler[0].'::'.$handler[1];
62+
}
63+
64+
if (\is_string($handler)) {
65+
return $handler;
66+
}
67+
68+
if ($handler instanceof \Closure) {
69+
$r = new \ReflectionFunction($handler);
70+
if (false !== strpos($r->name, '{closure}')) {
71+
return 'Closure';
72+
}
73+
if ($class = $r->getClosureScopeClass()) {
74+
return $class->name.'::'.$r->name;
75+
}
76+
77+
return $r->name;
78+
}
79+
80+
return \get_class($handler).'::__invoke';
81+
}
82+
}

‎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('from_transport')) {
86+
return true;
87+
}
88+
89+
return $received->getTransportName() === $expectedTransport;
90+
}
6391
}

‎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
}

‎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)