vendor/symfony/messenger/Middleware/DispatchAfterCurrentBusMiddleware.php line 68

  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\Middleware;
  14. use Symfony\Component\Messenger\Envelope;
  15. use Symfony\Component\Messenger\Exception\DelayedMessageHandlingException;
  16. use Symfony\Component\Messenger\Stamp\DispatchAfterCurrentBusStamp;
  18. /**
  19.  * Allow to configure messages to be handled after the current bus is finished.
  20.  *
  21.  * I.e, messages dispatched from a handler with a DispatchAfterCurrentBus stamp
  22.  * will actually be handled once the current message being dispatched is fully
  23.  * handled.
  24.  *
  25.  * For instance, using this middleware before the DoctrineTransactionMiddleware
  26.  * means sub-dispatched messages with a DispatchAfterCurrentBus stamp would be
  27.  * handled after the Doctrine transaction has been committed.
  28.  *
  29.  * @author Tobias Nyholm <tobias.nyholm@gmail.com>
  30.  */
  31. class DispatchAfterCurrentBusMiddleware implements MiddlewareInterface
  32. {
  33.     /**
  34.      * @var QueuedEnvelope[] A queue of messages and next middleware
  35.      */
  36.     private array $queue = [];
  38.     /**
  39.      * @var bool this property is used to signal if we are inside a the first/root call to
  40.      *           MessageBusInterface::dispatch() or if dispatch has been called inside a message handler
  41.      */
  42.     private bool $isRootDispatchCallRunning = false;
  44.     public function handle(Envelope $envelope, StackInterface $stack): Envelope
  45.     {
  46.         if (null !== $envelope->last(DispatchAfterCurrentBusStamp::class)) {
  47.             if ($this->isRootDispatchCallRunning) {
  48.                 $this->queue[] = new QueuedEnvelope($envelope, $stack);
  50.                 return $envelope;
  51.             }
  53.             $envelope = $envelope->withoutAll(DispatchAfterCurrentBusStamp::class);
  54.         }
  56.         if ($this->isRootDispatchCallRunning) {
  57.             /*
  58.              * A call to MessageBusInterface::dispatch() was made from inside the main bus handling,
  59.              * but the message does not have the stamp. So, process it like normal.
  60.              */
  61.             return $stack->next()->handle($envelope, $stack);
  62.         }
  64.         // First time we get here, mark as inside a "root dispatch" call:
  65.         $this->isRootDispatchCallRunning = true;
  66.         try {
  67.             // Execute the whole middleware stack & message handling for main dispatch:
  68.             $returnedEnvelope = $stack->next()->handle($envelope, $stack);
  69.         } catch (\Throwable $exception) {
  70.             /*
  71.              * Whenever an exception occurs while handling a message that has
  72.              * queued other messages, we drop the queued ones.
  73.              * This is intentional since the queued commands were likely dependent
  74.              * on the preceding command.
  75.              */
  76.             $this->queue = [];
  77.             $this->isRootDispatchCallRunning = false;
  79.             throw $exception;
  80.         }
  82.         // "Root dispatch" call is finished, dispatch stored messages.
  83.         $exceptions = [];
  84.         while (null !== $queueItem = array_shift($this->queue)) {
  85.             // Save how many messages are left in queue before handling the message
  86.             $queueLengthBefore = \count($this->queue);
  87.             try {
  88.                 // Execute the stored messages
  89.                 $queueItem->getStack()->next()->handle($queueItem->getEnvelope(), $queueItem->getStack());
  90.             } catch (\Exception $exception) {
  91.                 // Gather all exceptions
  92.                 $exceptions[] = $exception;
  93.                 // Restore queue to previous state
  94.                 $this->queue = \array_slice($this->queue, 0, $queueLengthBefore);
  95.             }
  96.         }
  98.         $this->isRootDispatchCallRunning = false;
  99.         if (\count($exceptions) > 0) {
  100.             throw new DelayedMessageHandlingException($exceptions);
  101.         }
  103.         return $returnedEnvelope;
  104.     }
  105. }
  107. /**
  108.  * @internal
  109.  */
  110. final class QueuedEnvelope
  111. {
  112.     private Envelope $envelope;
  113.     private StackInterface $stack;
  115.     public function __construct(Envelope $envelope, StackInterface $stack)
  116.     {
  117.         $this->envelope = $envelope->withoutAll(DispatchAfterCurrentBusStamp::class);
  118.         $this->stack = $stack;
  119.     }
  121.     public function getEnvelope(): Envelope
  122.     {
  123.         return $this->envelope;
  124.     }
  126.     public function getStack(): StackInterface
  127.     {
  128.         return $this->stack;
  129.     }
  130. }