[Messenger] make middlewares truly lazy on a bus
This commit is contained in:
parent
f24794da31
commit
6a5d7a1aac
@ -11,28 +11,39 @@
|
|||||||
|
|
||||||
namespace Symfony\Component\Messenger;
|
namespace Symfony\Component\Messenger;
|
||||||
|
|
||||||
use Symfony\Component\Messenger\Exception\InvalidArgumentException;
|
|
||||||
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
|
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @author Samuel Roze <samuel.roze@gmail.com>
|
* @author Samuel Roze <samuel.roze@gmail.com>
|
||||||
* @author Matthias Noback <matthiasnoback@gmail.com>
|
* @author Matthias Noback <matthiasnoback@gmail.com>
|
||||||
|
* @author Nicolas Grekas <p@tchwork.com>
|
||||||
*/
|
*/
|
||||||
class MessageBus implements MessageBusInterface
|
class MessageBus implements MessageBusInterface
|
||||||
{
|
{
|
||||||
private $middlewareHandlers;
|
private $middlewareAggregate;
|
||||||
|
|
||||||
/**
|
|
||||||
* @var MiddlewareInterface[]|null
|
|
||||||
*/
|
|
||||||
private $indexedMiddlewareHandlers;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param MiddlewareInterface[]|iterable $middlewareHandlers
|
* @param MiddlewareInterface[]|iterable $middlewareHandlers
|
||||||
*/
|
*/
|
||||||
public function __construct(iterable $middlewareHandlers = array())
|
public function __construct(iterable $middlewareHandlers = array())
|
||||||
{
|
{
|
||||||
$this->middlewareHandlers = $middlewareHandlers;
|
if ($middlewareHandlers instanceof \IteratorAggregate) {
|
||||||
|
$this->middlewareAggregate = $middlewareHandlers;
|
||||||
|
} elseif (\is_array($middlewareHandlers)) {
|
||||||
|
$this->middlewareAggregate = new \ArrayObject($middlewareHandlers);
|
||||||
|
} else {
|
||||||
|
$this->middlewareAggregate = new class() {
|
||||||
|
public $aggregate;
|
||||||
|
public $iterator;
|
||||||
|
|
||||||
|
public function getIterator()
|
||||||
|
{
|
||||||
|
return $this->aggregate = new \ArrayObject(iterator_to_array($this->iterator, false));
|
||||||
|
}
|
||||||
|
};
|
||||||
|
$this->middlewareAggregate->aggregate = &$this->middlewareAggregate;
|
||||||
|
$this->middlewareAggregate->iterator = $middlewareHandlers;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -41,24 +52,26 @@ class MessageBus implements MessageBusInterface
|
|||||||
public function dispatch($message): void
|
public function dispatch($message): void
|
||||||
{
|
{
|
||||||
if (!\is_object($message)) {
|
if (!\is_object($message)) {
|
||||||
throw new InvalidArgumentException(sprintf('Invalid type for message argument. Expected object, but got "%s".', \gettype($message)));
|
throw new \TypeError(sprintf('Invalid argument provided to "%s()": expected object, but got %s.', __METHOD__, \gettype($message)));
|
||||||
}
|
}
|
||||||
|
$middlewareIterator = $this->middlewareAggregate->getIterator();
|
||||||
|
|
||||||
$this->callableForNextMiddleware(0)($message instanceof Envelope ? $message : new Envelope($message));
|
while ($middlewareIterator instanceof \IteratorAggregate) {
|
||||||
|
$middlewareIterator = $middlewareIterator->getIterator();
|
||||||
}
|
}
|
||||||
|
$middlewareIterator->rewind();
|
||||||
|
|
||||||
private function callableForNextMiddleware(int $index): callable
|
if (!$middlewareIterator->valid()) {
|
||||||
{
|
return;
|
||||||
if (null === $this->indexedMiddlewareHandlers) {
|
|
||||||
$this->indexedMiddlewareHandlers = \is_array($this->middlewareHandlers) ? array_values($this->middlewareHandlers) : iterator_to_array($this->middlewareHandlers, false);
|
|
||||||
}
|
}
|
||||||
|
$next = static function (Envelope $envelope) use ($middlewareIterator, &$next) {
|
||||||
|
$middlewareIterator->next();
|
||||||
|
|
||||||
if (!isset($this->indexedMiddlewareHandlers[$index])) {
|
if ($middlewareIterator->valid()) {
|
||||||
return static function () {};
|
$middlewareIterator->current()->handle($envelope, $next);
|
||||||
}
|
}
|
||||||
|
|
||||||
return function (Envelope $envelope) use ($index) {
|
|
||||||
$this->indexedMiddlewareHandlers[$index]->handle($envelope, $this->callableForNextMiddleware($index + 1));
|
|
||||||
};
|
};
|
||||||
|
|
||||||
|
$middlewareIterator->current()->handle($message instanceof Envelope ? $message : new Envelope($message), $next);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -90,10 +90,10 @@ class MessengerPassTest extends TestCase
|
|||||||
->setAbstract(true)
|
->setAbstract(true)
|
||||||
;
|
;
|
||||||
|
|
||||||
$middlewares = array(array('id' => 'call_message_handler'));
|
$middlewareHandlers = array(array('id' => 'call_message_handler'));
|
||||||
|
|
||||||
$container->setParameter($commandBusId.'.middleware', $middlewares);
|
$container->setParameter($commandBusId.'.middleware', $middlewareHandlers);
|
||||||
$container->setParameter($queryBusId.'.middleware', $middlewares);
|
$container->setParameter($queryBusId.'.middleware', $middlewareHandlers);
|
||||||
|
|
||||||
$container->register(DummyCommandHandler::class)->addTag('messenger.message_handler', array('bus' => $commandBusId));
|
$container->register(DummyCommandHandler::class)->addTag('messenger.message_handler', array('bus' => $commandBusId));
|
||||||
$container->register(DummyQueryHandler::class)->addTag('messenger.message_handler', array('bus' => $queryBusId));
|
$container->register(DummyQueryHandler::class)->addTag('messenger.message_handler', array('bus' => $queryBusId));
|
||||||
@ -607,10 +607,10 @@ class MessengerPassTest extends TestCase
|
|||||||
|
|
||||||
$container->register('console.command.messenger_debug', DebugCommand::class)->addArgument(array());
|
$container->register('console.command.messenger_debug', DebugCommand::class)->addArgument(array());
|
||||||
|
|
||||||
$middlewares = array(array('id' => 'call_message_handler'));
|
$middlewareHandlers = array(array('id' => 'call_message_handler'));
|
||||||
|
|
||||||
$container->setParameter($commandBusId.'.middleware', $middlewares);
|
$container->setParameter($commandBusId.'.middleware', $middlewareHandlers);
|
||||||
$container->setParameter($queryBusId.'.middleware', $middlewares);
|
$container->setParameter($queryBusId.'.middleware', $middlewareHandlers);
|
||||||
|
|
||||||
$container->register(DummyCommandHandler::class)->addTag('messenger.message_handler', array('bus' => $commandBusId));
|
$container->register(DummyCommandHandler::class)->addTag('messenger.message_handler', array('bus' => $commandBusId));
|
||||||
$container->register(DummyQueryHandler::class)->addTag('messenger.message_handler', array('bus' => $queryBusId));
|
$container->register(DummyQueryHandler::class)->addTag('messenger.message_handler', array('bus' => $queryBusId));
|
||||||
|
@ -30,8 +30,8 @@ class MessageBusTest extends TestCase
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @expectedException \Symfony\Component\Messenger\Exception\InvalidArgumentException
|
* @expectedException \TypeError
|
||||||
* @expectedExceptionMessage Invalid type for message argument. Expected object, but got "string".
|
* @expectedExceptionMessage Invalid argument provided to "Symfony\Component\Messenger\MessageBus::dispatch()": expected object, but got string.
|
||||||
*/
|
*/
|
||||||
public function testItDispatchInvalidMessageType()
|
public function testItDispatchInvalidMessageType()
|
||||||
{
|
{
|
||||||
|
Reference in New Issue
Block a user