8000 [Messenger] allow specifying the dispatch key when calling dispatch() · symfony/symfony@292ec4a · GitHub
[go: up one dir, main page]
Skip to content

Commit 292ec4a

Browse files
nicolas-grekasnicolas-grekas
committed
[Messenger] allow specifying the dispatch key when calling dispatch()
1 parent 9aaec94 commit 292ec4a

File tree

14 files changed

+126
-16
lines changed

14 files changed

+126
-16
lines changed

‎src/Symfony/Bundle/FrameworkBundle/Controller/ControllerTrait.php

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
use Symfony\Component\HttpFoundation\StreamedResponse;
2525
use Symfony\Component\HttpKernel\Exception\NotFoundHttpException;
2626
use Symfony\Component\HttpKernel\HttpKernelInterface;
27+
use Symfony\Component\Messenger\Envelope;
2728
use Symfony\Component\Routing\Generator\UrlGeneratorInterface;
2829
use Symfony\Component\Security\Core\Exception\AccessDeniedException;
2930
use Symfony\Component\Security\Csrf\CsrfToken;
@@ -386,16 +387,18 @@ protected function isCsrfTokenValid(string $id, ?string $token): bool
386387
/**
387388
* Dispatches a message to the bus.
388389
*
389-
* @param object $message The message to dispatch
390+
* @param object|Envelope $message The message or the message pre-wrapped in an envelope
391+
* @param string|null $key The key to use to select the senders and/or the handlers for the message;
392+
* if not provided, the key is derived from the class of the message
390393
*
391394
* @final
392395
*/
393-
protected function dispatchMessage($message)
396+
protected function dispatchMessage($message, string $key = null): void
394397
{
395398
if (!$this->container->has('message_bus')) {
396399
throw new \LogicException('The message bus is not enabled in your application. Try running "composer require symfony/messenger".');
397400
}
398401

399-
return $this->container->get('message_bus')->dispatch($message);
402+
$this->container->get('message_bus')->dispatch($message, $key);
400403
}
401404
}

‎src/Symfony/Component/Messenger/Asynchronous/Routing/AbstractSenderLocator.php

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,12 @@ abstract class AbstractSenderLocator implements SenderLocatorInterface
2222
{
2323
public static function getValueFromMessageRouting(array $mapping, Envelope $envelope)
2424
{
25+
$name = $envelope->getDispatchKey();
26+
27+
if (null !== $name && isset($mapping[$name])) {
28+
return $mapping[$name];
29+
}
30+
2531
if (isset($mapping[$class = \get_class($envelope->getMessage())])) {
2632
return $mapping[$class];
2733
}

‎src/Symfony/Component/Messenger/Asynchronous/Routing/SenderLocator.php

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ public function getSender(Envelope $envelope): ?SenderInterface
3838
}
3939

4040
if (!$sender instanceof SenderInterface) {
41-
throw new RuntimeException(sprintf('The sender instance provided for message "%s" should be of type "%s" but got "%s".', \get_class($envelope->getMessage()), SenderInterface::class, \is_object($sender) ? \get_class($sender) : \gettype($sender)));
41+
throw new RuntimeException(sprintf('The sender instance provided for message "%s" should be of type "%s" but got "%s".', $envelope->getDispatchKey() ?? \get_class($envelope->getMessage()), SenderInterface::class, \is_object($sender) ? \get_class($sender) : \gettype($sender)));
4242
}
4343

4444
return $sender;

‎src/Symfony/Component/Messenger/CHANGELOG.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ CHANGELOG
66

77
* The component is not experimental anymore
88
* All the changes below are BC BREAKS
9+
* `MessageBusInterface::dispatch()` now requires a second `string $key = null` argument
910
* `MessageBusInterface::dispatch()` and `MiddlewareInterface::handle()` now return `void`
1011
* `MiddlewareInterface::handle()` now require an `Envelope` as first argument
1112
* `EnvelopeAwareInterface` has been removed

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

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111

1212
namespace Symfony\Component\Messenger;
1313

14+
use Symfony\Component\Messenger\Stamp\DispatchKeyStamp;
1415
use Symfony\Component\Messenger\Stamp\StampInterface;
1516

1617
/**
@@ -38,6 +39,18 @@ public function __construct($message, StampInterface ...$stamps)
3839
}
3940
}
4041

42+
/**
43+
* Wraps a message into an envelope if not already wrapped.
44+
*
45+
* @param Envelope|object $message
46+
*/
47+
public static function wrap($message, ?string $key): self
48+
{
49+
$envelope = $message instanceof self ? $message : new self($message);
50+
51+
return null !== $key ? $envelope->with(new DispatchKeyStamp($key)) : $envelope;
52+
}
53+
4154
/**
4255
* @return Envelope a new Envelope instance with additional stamp
4356
*/
@@ -72,4 +85,13 @@ public function getMessage()
7285
{
7386
return $this->message;
7487
}
88+
89+
public function getDispatchKey(): ?string
90+
{
91+
if (null !== $key = $this->stamps[DispatchKeyStamp::class] ?? null) {
92+
$key = $key->getName();
93+
}
94+
95+
return $key;
96+
}
7597
}

‎src/Symfony/Component/Messenger/Handler/Locator/AbstractHandlerLocator.php

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,12 @@ abstract class AbstractHandlerLocator implements HandlerLocatorInterface
2424
{
2525
public function getHandler(Envelope $envelope): callable
2626
{
27+
$name = $envelope->getDispatchKey();
28+
29+
if (null !== $name && $handler = $this->getHandlerByName($name)) {
30+
return $handler;
31+
}
32+
2733
$class = \get_class($envelope->getMessage());
2834

2935
if ($handler = $this->getHandlerByName($class)) {

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ public function getIterator()
4949
/**
5050
* {@inheritdoc}
5151
*/
52-
public function dispatch($message): void
52+
public function dispatch($message, string $key = null): void
5353
{
5454
if (!\is_object($message)) {
5555
throw new \TypeError(sprintf('Invalid argument provided to "%s()": expected object, but got %s.', __METHOD__, \gettype($message)));
@@ -72,6 +72,6 @@ public function dispatch($message): void
7272
}
7373
};
7474

75-
$middlewareIterator->current()->handle($message instanceof Envelope ? $message : new Envelope($message), $next);
75+
$middlewareIterator->current()->handle(Envelope::wrap($message, $key), $next);
7676
}
7777
}

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@ interface MessageBusInterface
2020
* Dispatches the given message.
2121
*
2222
* @param object|Envelope $message The message or the message pre-wrapped in an envelope
23+
* @param string|null $key The key to use to select the senders and/or the handlers for the message;
24+
* if not provided, the key is derived from the class of the message
2325
*/
24-
public function dispatch($message): void;
26+
public function dispatch($message, string $key = null): void;
2527
}

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,19 +34,19 @@ public function handle(Envelope $envelope, callable $next): void
3434
$message = $envelope->getMessage();
3535
$context = array(
3636
'message' => $message,
37-
'name' => \get_class($message),
37+
'name' => $envelope->getDispatchKey() ?? \get_class($message),
3838
);
39-
$this->logger->debug('Starting handling message {name}', $context);
39+
$this->logger->debug('Starting handling message "{name}"', $context);
4040

4141
try {
4242
$next($envelope);
4343
} catch (\Throwable $e) {
4444
$context['exception'] = $e;
45-
$this->logger->warning('An exception occurred while handling message {name}', $context);
45+
$this->logger->warning('An exception occurred while handling message "{name}"', $context);
4646

4747
throw $e;
4848
}
4949

50-
$this->logger->debug('Finished handling message {name}', $context);
50+
$this->logger->debug('Finished handling message "{name}"', $context);
5151
}
5252
}

‎src/Symfony/Component/Messenger/Stamp/DispatchKeyStamp.php

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
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\Stamp;
13+
14+
/**
15+
* A stamp that defines the key that should be used to decide
16+
* which senders and/or handlers are bound to a message.
17+
*
18+
* @author Nicolas Grekas <p@tchwork.com>
19+
*/
20+
final class DispatchKeyStamp implements StampInterface
21+
{
22+
private $name;
23+
24+
public function __construct(string $name)
25+
{
26+
$this->name = $name;
27+
}
28+
29+
public function getName()
30+
{
31+
return $this->name;
32+
}
33+
}

‎src/Symfony/Component/Messenger/Tests/Asynchronous/Routing/ContainerSenderLocatorTest.php

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,24 @@ public function testItReturnsTheSenderBasedOnTheMessageInterface()
7171
$this->assertNull($locator->getSender(new Envelope(new SecondMessage())));
7272
}
7373

74+
public function testItReturnsTheSenderBasedOnTheDispatchKey()
75+
{
76+
$container = new Container();
77+
$container->set('my_amqp_sender1', $sender1 = $this->getMockBuilder(SenderInterface::class)->getMock());
78+
$container->set('my_amqp_sender2', $sender2 = $this->getMockBuilder(SenderInterface::class)->getMock());
79+
80+
$locator = new ContainerSenderLocator($container, array(
81+
DummyMessageInterface::class => 'my_amqp_sender1',
82+
));
83+
$this->assertSame($sender1, $locator->getSender(Envelope::wrap(new DummyMessage('Hello'), 'foo')));
84+
85+
$locator = new ContainerSenderLocator($container, array(
86+
DummyMessageInterface::class => 'my_amqp_sender1',
87+
'foo' => 'my_amqp_sender2',
88+
));
89+
$this->assertSame($sender2, $locator->getSender(Envelope::wrap(new DummyMessage('Hello'), 'foo')));
90+
}
91+
7492
public function testItSupportsAWildcardInsteadOfTheMessageClass()
7593
{
7694
$container = new Container();

‎src/Symfony/Component/Messenger/Tests/Handler/Locator/ContainerHandlerLocatorTest.php

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,4 +60,21 @@ public function testGetHandlerViaParentClass()
6060

6161
$this->assertSame($handler, $resolvedHandler);
6262
}
63+
64+
public function testGetHandlerViaDispatchKey()
65+
{
66+
$handler1 = function () {};
67+
$handler2 = function () {};
68+
69+
$container = new Container();
70+
$locator = new ContainerHandlerLocator($container);
71+
$container->set(DummyMessage::class, $handler1);
72+
73+
$resolvedHandler = $locator->getHandler(Envelope::wrap(new DummyMessage('Hey'), 'foo'));
74+
$this->assertSame($handler1, $resolvedHandler);
75+
76+
$container->set('foo', $handler2);
77+
$resolvedHandler = $locator->getHandler(Envelope::wrap(new DummyMessage('Hey'), 'foo'));
78+
$this->assertSame($handler2, $resolvedHandler);
79+
}
6380
}

‎src/Symfony/Component/Messenger/Tests/Transport/AmqpExt/Fixtures/long_receiver.php

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212

1313
require_once $autoload;
1414

15+
use Symfony\Component\Messenger\Envelope;
1516
use Symfony\Component\Messenger\MessageBusInterface;
1617
use Symfony\Component\Messenger\Transport\AmqpExt\AmqpReceiver;
1718
use Symfony\Component\Messenger\Transport\AmqpExt\Connection;
@@ -29,9 +30,10 @@
2930
$receiver = new AmqpReceiver($connection, $serializer);
3031

3132
$worker = new Worker($receiver, new class() implements MessageBusInterface {
32-
public function dispatch($envelope): void
33+
public function dispatch($message, string $key = null): void
3334
{
34-
echo 'Get envelope with message: '.get_class($envelope->getMessage())."\n";
35+
$envelope = Envelope::wrap($message, $key);
36+
echo 'Get envelope with message: '.($envelope->getDispatchKey() ?? \get_class($envelope->getMessage()))."\n";
3537
echo sprintf("with stamps: %s\n", json_encode(array_keys($envelope->all()), JSON_PRETTY_PRINT));
3638

3739
sleep(30);

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,9 +27,9 @@ public function __construct(MessageBusInterface $decoratedBus)
2727
/**
2828
* {@inheritdoc}
2929
*/
30-
public function dispatch($message): void
30+
public function dispatch($message, string $key = null): void
3131
{
32-
$envelope = $message instanceof Envelope ? $message : new Envelope($message);
32+
$envelope = Envelope::wrap($message, $key);
3333
$context = array(
3434
'stamps' => array_values($envelope->all()),
3535
'message' => $envelope->getMessage(),
@@ -38,7 +38,7 @@ public function dispatch($message): void
3838
);
3939

4040
try {
41-
$this->decoratedBus->dispatch($message);
41+
$this->decoratedBus->dispatch($message, $key);
4242
} catch (\Throwable $e) {
4343
$context['exception'] = $e;
4444

0 commit comments

Comments
 (0)
0