vendor/symfony/messenger/EventListener/SendFailedMessageToFailureTransportListener.php line 46

Open in your IDE?
  1. <?php
  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. */
  12. namespace Symfony\Component\Messenger\EventListener;
  14. use Psr\Container\ContainerInterface;
  15. use Psr\Log\LoggerInterface;
  16. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  17. use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
  18. use Symfony\Component\Messenger\Stamp\DelayStamp;
  19. use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
  20. use Symfony\Component\Messenger\Stamp\SentToFailureTransportStamp;
  21. use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
  23. /**
  24. * Sends a rejected message to a "failure transport".
  25. *
  26. * @author Ryan Weaver <ryan@symfonycasts.com>
  27. */
  28. class SendFailedMessageToFailureTransportListener implements EventSubscriberInterface
  29. {
  30. private $failureSenders;
  31. private $logger;
  33. /**
  34. * @param ContainerInterface $failureSenders
  35. */
  36. public function __construct($failureSenders, ?LoggerInterface $logger = null)
  37. {
  38. if (!$failureSenders instanceof ContainerInterface) {
  39. trigger_deprecation('symfony/messenger', '5.3', 'Passing a SenderInterface value as 1st argument to "%s()" is deprecated, pass a ServiceLocator instead.', __METHOD__);
  40. }
  42. $this->failureSenders = $failureSenders;
  43. $this->logger = $logger;
  44. }
  46. public function onMessageFailed(WorkerMessageFailedEvent $event)
  47. {
  48. if ($event->willRetry()) {
  49. return;
  50. }
  52. if (!$this->hasFailureTransports($event)) {
  53. return;
  54. }
  56. $failureSender = $this->getFailureSender($event->getReceiverName());
  57. if (null === $failureSender) {
  58. return;
  59. }
  61. $envelope = $event->getEnvelope();
  63. // avoid re-sending to the failed sender
  64. if (null !== $envelope->last(SentToFailureTransportStamp::class)) {
  65. return;
  66. }
  68. $envelope = $envelope->with(
  69. new SentToFailureTransportStamp($event->getReceiverName()),
  70. new DelayStamp(0),
  71. new RedeliveryStamp(0)
  72. );
  74. if (null !== $this->logger) {
  75. $this->logger->info('Rejected message {class} will be sent to the failure transport {transport}.', [
  76. 'class' => \get_class($envelope->getMessage()),
  77. 'transport' => \get_class($failureSender),
  78. ]);
  79. }
  81. $failureSender->send($envelope);
  82. }
  84. public static function getSubscribedEvents()
  85. {
  86. return [
  87. WorkerMessageFailedEvent::class => ['onMessageFailed', -100],
  88. ];
  89. }
  91. private function getFailureSender(string $receiverName): SenderInterface
  92. {
  93. if ($this->failureSenders instanceof SenderInterface) {
  94. return $this->failureSenders;
  95. }
  97. return $this->failureSenders->get($receiverName);
  98. }
  100. private function hasFailureTransports(WorkerMessageFailedEvent $event): bool
  101. {
  102. return ($this->failureSenders instanceof ContainerInterface && $this->failureSenders->has($event->getReceiverName())) || $this->failureSenders instanceof SenderInterface;
  103. }
  104. }