vendor/symfony/messenger/Middleware/HandleMessageMiddleware.php line 72

  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 Psr\Log\LoggerAwareTrait;
  15. use Psr\Log\NullLogger;
  16. use Symfony\Component\Messenger\Envelope;
  17. use Symfony\Component\Messenger\Exception\HandlerFailedException;
  18. use Symfony\Component\Messenger\Exception\LogicException;
  19. use Symfony\Component\Messenger\Exception\NoHandlerForMessageException;
  20. use Symfony\Component\Messenger\Handler\Acknowledger;
  21. use Symfony\Component\Messenger\Handler\HandlerDescriptor;
  22. use Symfony\Component\Messenger\Handler\HandlersLocatorInterface;
  23. use Symfony\Component\Messenger\Stamp\AckStamp;
  24. use Symfony\Component\Messenger\Stamp\FlushBatchHandlersStamp;
  25. use Symfony\Component\Messenger\Stamp\HandledStamp;
  26. use Symfony\Component\Messenger\Stamp\HandlerArgumentsStamp;
  27. use Symfony\Component\Messenger\Stamp\NoAutoAckStamp;
  29. /**
  30.  * @author Samuel Roze <samuel.roze@gmail.com>
  31.  */
  32. class HandleMessageMiddleware implements MiddlewareInterface
  33. {
  34.     use LoggerAwareTrait;
  36.     private HandlersLocatorInterface $handlersLocator;
  37.     private bool $allowNoHandlers;
  39.     public function __construct(HandlersLocatorInterface $handlersLocator, bool $allowNoHandlers = false)
  40.     {
  41.         $this->handlersLocator = $handlersLocator;
  42.         $this->allowNoHandlers = $allowNoHandlers;
  43.         $this->logger = new NullLogger();
  44.     }
  46.     /**
  47.      * @throws NoHandlerForMessageException When no handler is found and $allowNoHandlers is false
  48.      */
  49.     public function handle(Envelope $envelope, StackInterface $stack): Envelope
  50.     {
  51.         $handler = null;
  52.         $message = $envelope->getMessage();
  54.         $context = [
  55.             'class' => $message::class,
  56.         ];
  58.         $exceptions = [];
  59.         $alreadyHandled = false;
  60.         foreach ($this->handlersLocator->getHandlers($envelope) as $handlerDescriptor) {
  61.             if ($this->messageHasAlreadyBeenHandled($envelope, $handlerDescriptor)) {
  62.                 $alreadyHandled = true;
  63.                 continue;
  64.             }
  66.             try {
  67.                 $handler = $handlerDescriptor->getHandler();
  68.                 $batchHandler = $handlerDescriptor->getBatchHandler();
  70.                 /** @var AckStamp $ackStamp */
  71.                 if ($batchHandler && $ackStamp = $envelope->last(AckStamp::class)) {
  72.                     $ack = new Acknowledger(get_debug_type($batchHandler), static function (\Throwable $e = null, $result = null) use ($envelope, $ackStamp, $handlerDescriptor) {
  73.                         if (null !== $e) {
  74.                             $e = new HandlerFailedException($envelope, [$e]);
  75.                         } else {
  76.                             $envelope = $envelope->with(HandledStamp::fromDescriptor($handlerDescriptor, $result));
  77.                         }
  79.                         $ackStamp->ack($envelope, $e);
  80.                     });
  82.                     $result = $this->callHandler($handler, $message, $ack, $envelope->last(HandlerArgumentsStamp::class));
  84.                     if (!\is_int($result) || 0 > $result) {
  85.                         throw new LogicException(sprintf('A handler implementing BatchHandlerInterface must return the size of the current batch as a positive integer, "%s" returned from "%s".', \is_int($result) ? $result : get_debug_type($result), get_debug_type($batchHandler)));
  86.                     }
  88.                     if (!$ack->isAcknowledged()) {
  89.                         $envelope = $envelope->with(new NoAutoAckStamp($handlerDescriptor));
  90.                     } elseif ($ack->getError()) {
  91.                         throw $ack->getError();
  92.                     } else {
  93.                         $result = $ack->getResult();
  94.                     }
  95.                 } else {
  96.                     $result = $this->callHandler($handler, $message, null, $envelope->last(HandlerArgumentsStamp::class));
  97.                 }
  99.                 $handledStamp = HandledStamp::fromDescriptor($handlerDescriptor, $result);
  100.                 $envelope = $envelope->with($handledStamp);
  101.                 $this->logger->info('Message {class} handled by {handler}', $context + ['handler' => $handledStamp->getHandlerName()]);
  102.             } catch (\Throwable $e) {
  103.                 $exceptions[] = $e;
  104.             }
  105.         }
  107.         /** @var FlushBatchHandlersStamp $flushStamp */
  108.         if ($flushStamp = $envelope->last(FlushBatchHandlersStamp::class)) {
  109.             /** @var NoAutoAckStamp $stamp */
  110.             foreach ($envelope->all(NoAutoAckStamp::class) as $stamp) {
  111.                 try {
  112.                     $handler = $stamp->getHandlerDescriptor()->getBatchHandler();
  113.                     $handler->flush($flushStamp->force());
  114.                 } catch (\Throwable $e) {
  115.                     $exceptions[] = $e;
  116.                 }
  117.             }
  118.         }
  120.         if (null === $handler && !$alreadyHandled) {
  121.             if (!$this->allowNoHandlers) {
  122.                 throw new NoHandlerForMessageException(sprintf('No handler for message "%s".', $context['class']));
  123.             }
  125.             $this->logger->info('No handler for message {class}', $context);
  126.         }
  128.         if (\count($exceptions)) {
  129.             throw new HandlerFailedException($envelope, $exceptions);
  130.         }
  132.         return $stack->next()->handle($envelope, $stack);
  133.     }
  135.     private function messageHasAlreadyBeenHandled(Envelope $envelope, HandlerDescriptor $handlerDescriptor): bool
  136.     {
  137.         /** @var HandledStamp $stamp */
  138.         foreach ($envelope->all(HandledStamp::class) as $stamp) {
  139.             if ($stamp->getHandlerName() === $handlerDescriptor->getName()) {
  140.                 return true;
  141.             }
  142.         }
  144.         return false;
  145.     }
  147.     private function callHandler(callable $handler, object $message, ?Acknowledger $ack, ?HandlerArgumentsStamp $handlerArgumentsStamp): mixed
  148.     {
  149.         $arguments = [$message];
  150.         if (null !== $ack) {
  151.             $arguments[] = $ack;
  152.         }
  153.         if (null !== $handlerArgumentsStamp) {
  154.             $arguments = [...$arguments, ...$handlerArgumentsStamp->getAdditionalArguments()];
  155.         }
  157.         return $handler(...$arguments);
  158.     }
  159. }