MessageBus.php 2.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879
  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;
  11. use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
  12. use Symfony\Component\Messenger\Middleware\StackMiddleware;
  13. /**
  14. * @author Samuel Roze <samuel.roze@gmail.com>
  15. * @author Matthias Noback <matthiasnoback@gmail.com>
  16. * @author Nicolas Grekas <p@tchwork.com>
  17. */
  18. class MessageBus implements MessageBusInterface
  19. {
  20. private $middlewareAggregate;
  21. /**
  22. * @param iterable<mixed, MiddlewareInterface> $middlewareHandlers
  23. */
  24. public function __construct(iterable $middlewareHandlers = [])
  25. {
  26. if ($middlewareHandlers instanceof \IteratorAggregate) {
  27. $this->middlewareAggregate = $middlewareHandlers;
  28. } elseif (\is_array($middlewareHandlers)) {
  29. $this->middlewareAggregate = new \ArrayObject($middlewareHandlers);
  30. } else {
  31. // $this->middlewareAggregate should be an instance of IteratorAggregate.
  32. // When $middlewareHandlers is an Iterator, we wrap it to ensure it is lazy-loaded and can be rewound.
  33. $this->middlewareAggregate = new class($middlewareHandlers) implements \IteratorAggregate {
  34. private $middlewareHandlers;
  35. private $cachedIterator;
  36. public function __construct(\Traversable $middlewareHandlers)
  37. {
  38. $this->middlewareHandlers = $middlewareHandlers;
  39. }
  40. public function getIterator(): \Traversable
  41. {
  42. if (null === $this->cachedIterator) {
  43. $this->cachedIterator = new \ArrayObject(iterator_to_array($this->middlewareHandlers, false));
  44. }
  45. return $this->cachedIterator;
  46. }
  47. };
  48. }
  49. }
  50. /**
  51. * {@inheritdoc}
  52. */
  53. public function dispatch(object $message, array $stamps = []): Envelope
  54. {
  55. $envelope = Envelope::wrap($message, $stamps);
  56. $middlewareIterator = $this->middlewareAggregate->getIterator();
  57. while ($middlewareIterator instanceof \IteratorAggregate) {
  58. $middlewareIterator = $middlewareIterator->getIterator();
  59. }
  60. $middlewareIterator->rewind();
  61. if (!$middlewareIterator->valid()) {
  62. return $envelope;
  63. }
  64. $stack = new StackMiddleware($middlewareIterator);
  65. return $middlewareIterator->current()->handle($envelope, $stack);
  66. }
  67. }