RoutableMessageBus.php 2.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768
  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 Psr\Container\ContainerInterface;
  12. use Symfony\Component\Messenger\Exception\InvalidArgumentException;
  13. use Symfony\Component\Messenger\Stamp\BusNameStamp;
  14. /**
  15. * Bus of buses that is routable using a BusNameStamp.
  16. *
  17. * This is useful when passed to Worker: messages received
  18. * from the transport can be sent to the correct bus.
  19. *
  20. * @author Ryan Weaver <ryan@symfonycasts.com>
  21. */
  22. class RoutableMessageBus implements MessageBusInterface
  23. {
  24. private $busLocator;
  25. private $fallbackBus;
  26. public function __construct(ContainerInterface $busLocator, MessageBusInterface $fallbackBus = null)
  27. {
  28. $this->busLocator = $busLocator;
  29. $this->fallbackBus = $fallbackBus;
  30. }
  31. public function dispatch($envelope, array $stamps = []): Envelope
  32. {
  33. if (!$envelope instanceof Envelope) {
  34. throw new InvalidArgumentException('Messages passed to RoutableMessageBus::dispatch() must be inside an Envelope.');
  35. }
  36. /** @var BusNameStamp|null $busNameStamp */
  37. $busNameStamp = $envelope->last(BusNameStamp::class);
  38. if (null === $busNameStamp) {
  39. if (null === $this->fallbackBus) {
  40. throw new InvalidArgumentException('Envelope is missing a BusNameStamp and no fallback message bus is configured on RoutableMessageBus.');
  41. }
  42. return $this->fallbackBus->dispatch($envelope, $stamps);
  43. }
  44. return $this->getMessageBus($busNameStamp->getBusName())->dispatch($envelope, $stamps);
  45. }
  46. /**
  47. * @internal
  48. */
  49. public function getMessageBus(string $busName): MessageBusInterface
  50. {
  51. if (!$this->busLocator->has($busName)) {
  52. throw new InvalidArgumentException(sprintf('Bus named "%s" does not exist.', $busName));
  53. }
  54. return $this->busLocator->get($busName);
  55. }
  56. }