vendor/symfony/messenger/Worker.php line 173

  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;
  14. use Psr\EventDispatcher\EventDispatcherInterface;
  15. use Psr\Log\LoggerInterface;
  16. use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
  17. use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
  18. use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
  19. use Symfony\Component\Messenger\Event\WorkerRateLimitedEvent;
  20. use Symfony\Component\Messenger\Event\WorkerRunningEvent;
  21. use Symfony\Component\Messenger\Event\WorkerStartedEvent;
  22. use Symfony\Component\Messenger\Event\WorkerStoppedEvent;
  23. use Symfony\Component\Messenger\Exception\HandlerFailedException;
  24. use Symfony\Component\Messenger\Exception\RejectRedeliveredMessageException;
  25. use Symfony\Component\Messenger\Exception\RuntimeException;
  26. use Symfony\Component\Messenger\Stamp\AckStamp;
  27. use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
  28. use Symfony\Component\Messenger\Stamp\FlushBatchHandlersStamp;
  29. use Symfony\Component\Messenger\Stamp\NoAutoAckStamp;
  30. use Symfony\Component\Messenger\Stamp\ReceivedStamp;
  31. use Symfony\Component\Messenger\Transport\Receiver\QueueReceiverInterface;
  32. use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
  33. use Symfony\Component\RateLimiter\LimiterInterface;
  35. /**
  36.  * @author Samuel Roze <samuel.roze@gmail.com>
  37.  * @author Tobias Schultze <http://tobion.de>
  38.  *
  39.  * @final
  40.  */
  41. class Worker
  42. {
  43.     private array $receivers;
  44.     private MessageBusInterface $bus;
  45.     private ?EventDispatcherInterface $eventDispatcher;
  46.     private ?LoggerInterface $logger;
  47.     private bool $shouldStop = false;
  48.     private WorkerMetadata $metadata;
  49.     private array $acks = [];
  50.     private \SplObjectStorage $unacks;
  51.     private ?array $rateLimiters;
  53.     /**
  54.      * @param ReceiverInterface[] $receivers Where the key is the transport name
  55.      */
  56.     public function __construct(array $receivers, MessageBusInterface $bus, EventDispatcherInterface $eventDispatcher = null, LoggerInterface $logger = null, array $rateLimiters = null)
  57.     {
  58.         $this->receivers = $receivers;
  59.         $this->bus = $bus;
  60.         $this->logger = $logger;
  61.         $this->eventDispatcher = $eventDispatcher;
  62.         $this->metadata = new WorkerMetadata([
  63.             'transportNames' => array_keys($receivers),
  64.         ]);
  65.         $this->unacks = new \SplObjectStorage();
  66.         $this->rateLimiters = $rateLimiters;
  67.     }
  69.     /**
  70.      * Receive the messages and dispatch them to the bus.
  71.      *
  72.      * Valid options are:
  73.      *  * sleep (default: 1000000): Time in microseconds to sleep after no messages are found
  74.      *  * queues: The queue names to consume from, instead of consuming from all queues. When this is used, all receivers must implement the QueueReceiverInterface
  75.      */
  76.     public function run(array $options = []): void
  77.     {
  78.         $options = array_merge([
  79.             'sleep' => 1000000,
  80.         ], $options);
  81.         $queueNames = $options['queues'] ?? null;
  83.         $this->metadata->set(['queueNames' => $queueNames]);
  85.         $this->eventDispatcher?->dispatch(new WorkerStartedEvent($this));
  87.         if ($queueNames) {
  88.             // if queue names are specified, all receivers must implement the QueueReceiverInterface
  89.             foreach ($this->receivers as $transportName => $receiver) {
  90.                 if (!$receiver instanceof QueueReceiverInterface) {
  91.                     throw new RuntimeException(sprintf('Receiver for "%s" does not implement "%s".', $transportName, QueueReceiverInterface::class));
  92.                 }
  93.             }
  94.         }
  96.         while (!$this->shouldStop) {
  97.             $envelopeHandled = false;
  98.             $envelopeHandledStart = microtime(true);
  99.             foreach ($this->receivers as $transportName => $receiver) {
  100.                 if ($queueNames) {
  101.                     $envelopes = $receiver->getFromQueues($queueNames);
  102.                 } else {
  103.                     $envelopes = $receiver->get();
  104.                 }
  106.                 foreach ($envelopes as $envelope) {
  107.                     $envelopeHandled = true;
  109.                     $this->rateLimit($transportName);
  110.                     $this->handleMessage($envelope, $transportName);
  111.                     $this->eventDispatcher?->dispatch(new WorkerRunningEvent($this, false));
  113.                     if ($this->shouldStop) {
  114.                         break 2;
  115.                     }
  116.                 }
  118.                 // after handling a single receiver, quit and start the loop again
  119.                 // this should prevent multiple lower priority receivers from
  120.                 // blocking too long before the higher priority are checked
  121.                 if ($envelopeHandled) {
  122.                     break;
  123.                 }
  124.             }
  126.             if (!$envelopeHandled && $this->flush(false)) {
  127.                 continue;
  128.             }
  130.             if (!$envelopeHandled) {
  131.                 $this->eventDispatcher?->dispatch(new WorkerRunningEvent($this, true));
  133.                 if (0 < $sleep = (int) ($options['sleep'] - 1e6 * (microtime(true) - $envelopeHandledStart))) {
  134.                     usleep($sleep);
  135.                 }
  136.             }
  137.         }
  139.         $this->flush(true);
  140.         $this->eventDispatcher?->dispatch(new WorkerStoppedEvent($this));
  141.     }
  143.     private function handleMessage(Envelope $envelope, string $transportName): void
  144.     {
  145.         $event = new WorkerMessageReceivedEvent($envelope, $transportName);
  146.         $this->eventDispatcher?->dispatch($event);
  147.         $envelope = $event->getEnvelope();
  149.         if (!$event->shouldHandle()) {
  150.             return;
  151.         }
  153.         $acked = false;
  154.         $ack = function (Envelope $envelope, \Throwable $e = null) use ($transportName, &$acked) {
  155.             $acked = true;
  156.             $this->acks[] = [$transportName, $envelope, $e];
  157.         };
  159.         try {
  160.             $e = null;
  161.             $envelope = $this->bus->dispatch($envelope->with(new ReceivedStamp($transportName), new ConsumedByWorkerStamp(), new AckStamp($ack)));
  162.         } catch (\Throwable $e) {
  163.         }
  165.         $noAutoAckStamp = $envelope->last(NoAutoAckStamp::class);
  167.         if (!$acked && !$noAutoAckStamp) {
  168.             $this->acks[] = [$transportName, $envelope, $e];
  169.         } elseif ($noAutoAckStamp) {
  170.             $this->unacks[$noAutoAckStamp->getHandlerDescriptor()->getBatchHandler()] = [$envelope->withoutAll(AckStamp::class), $transportName];
  171.         }
  173.         $this->ack();
  174.     }
  176.     private function ack(): bool
  177.     {
  178.         $acks = $this->acks;
  179.         $this->acks = [];
  181.         foreach ($acks as [$transportName, $envelope, $e]) {
  182.             $receiver = $this->receivers[$transportName];
  184.             if (null !== $e) {
  185.                 if ($rejectFirst = $e instanceof RejectRedeliveredMessageException) {
  186.                     // redelivered messages are rejected first so that continuous failures in an event listener or while
  187.                     // publishing for retry does not cause infinite redelivery loops
  188.                     $receiver->reject($envelope);
  189.                 }
  191.                 if ($e instanceof HandlerFailedException) {
  192.                     $envelope = $e->getEnvelope();
  193.                 }
  195.                 $failedEvent = new WorkerMessageFailedEvent($envelope, $transportName, $e);
  197.                 $this->eventDispatcher?->dispatch($failedEvent);
  198.                 $envelope = $failedEvent->getEnvelope();
  200.                 if (!$rejectFirst) {
  201.                     $receiver->reject($envelope);
  202.                 }
  204.                 continue;
  205.             }
  207.             $handledEvent = new WorkerMessageHandledEvent($envelope, $transportName);
  208.             $this->eventDispatcher?->dispatch($handledEvent);
  209.             $envelope = $handledEvent->getEnvelope();
  211.             if (null !== $this->logger) {
  212.                 $message = $envelope->getMessage();
  213.                 $context = [
  214.                     'class' => $message::class,
  215.                 ];
  216.                 $this->logger->info('{class} was handled successfully (acknowledging to transport).', $context);
  217.             }
  219.             $receiver->ack($envelope);
  220.         }
  222.         return (bool) $acks;
  223.     }
  225.     private function rateLimit(string $transportName): void
  226.     {
  227.         if (!$this->rateLimiters) {
  228.             return;
  229.         }
  231.         if (!\array_key_exists($transportName, $this->rateLimiters)) {
  232.             return;
  233.         }
  235.         /** @var LimiterInterface $rateLimiter */
  236.         $rateLimiter = $this->rateLimiters[$transportName]->create();
  237.         if ($rateLimiter->consume()->isAccepted()) {
  238.             return;
  239.         }
  241.         $this->logger?->info('Transport {transport} is being rate limited, waiting for token to become available...', ['transport' => $transportName]);
  243.         $this->eventDispatcher?->dispatch(new WorkerRateLimitedEvent($rateLimiter, $transportName));
  244.         $rateLimiter->reserve()->wait();
  245.     }
  247.     private function flush(bool $force): bool
  248.     {
  249.         $unacks = $this->unacks;
  251.         if (!$unacks->count()) {
  252.             return false;
  253.         }
  255.         $this->unacks = new \SplObjectStorage();
  257.         foreach ($unacks as $batchHandler) {
  258.             [$envelope, $transportName] = $unacks[$batchHandler];
  259.             try {
  260.                 $this->bus->dispatch($envelope->with(new FlushBatchHandlersStamp($force)));
  261.                 $envelope = $envelope->withoutAll(NoAutoAckStamp::class);
  262.                 unset($unacks[$batchHandler], $batchHandler);
  263.             } catch (\Throwable $e) {
  264.                 $this->acks[] = [$transportName, $envelope, $e];
  265.             }
  266.         }
  268.         return $this->ack();
  269.     }
  271.     public function stop(): void
  272.     {
  273.         $this->logger?->info('Stopping worker.', ['transport_names' => $this->metadata->getTransportNames()]);
  275.         $this->shouldStop = true;
  276.     }
  278.     public function getMetadata(): WorkerMetadata
  279.     {
  280.         return $this->metadata;
  281.     }
  282. }