Adding global retry support, events & more to messenger transport

Co-authored-by: Samuel ROZE <[email protected]>

Author: weaverryan

CHANGELOG.md

@@ -3,7 +3,34 @@ CHANGELOG
 
 4.3.0
 -----
-
+ 
+ * [BC BREAK] 2 new methods were added to `ReceiverInterface`:
+   `ack()` and `reject()`.
+ * [BC BREAK] Error handling was moved from the receivers into
+   `Worker`. Implementations of `ReceiverInterface::handle()`
+   should now allow all exceptions to be thrown, except for transport
+   exceptions. They should also not retry (e.g. if there's a queue,
+   remove from the queue) if there is a problem decoding the message.
+ * [BC BREAK] `RejectMessageExceptionInterface` was removed and replaced
+   by `Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException`,
+   which has the same behavior: a message will not be retried
+ * The default command name for `ConsumeMessagesCommand` was
+   changed from `messenger:consume-messages` to `messenger:consume`
+ * `ConsumeMessagesCommand` has two new optional constructor arguments
+ * `Worker` has 4 new option constructor arguments.
+ * The `Worker` class now handles calling `pcntl_signal_dispatch()` the
+   receiver no longer needs to call this.
+ * The `AmqpSender` will now retry messages using a dead-letter exchange
+   and delayed queues, instead of retrying via `nack()`
+ * Senders now receive the `Envelope` with the `SentStamp` on it. Previously,
+   the `Envelope` was passed to the sender and *then* the `SentStamp`
+   was added.
+ * `SerializerInterface` implementations should now throw a
+   `Symfony\Component\Messenger\Exception\MessageDecodingFailedException`
+   if `decode()` fails for any reason.
+ * [BC BREAK] The default `Serializer` will now throw a
+   `MessageDecodingFailedException` if `decode()` fails, instead
+   of the underlying exceptions from the Serializer component.
  * Added `PhpSerializer` which uses PHP's native `serialize()` and
    `unserialize()` to serialize messages to a transport
  * [BC BREAK] If no serializer were passed, the default serializer

Command/ConsumeMessagesCommand.php

@@ -20,6 +20,7 @@
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 use Symfony\Component\Messenger\Transport\Receiver\StopWhenMemoryUsageIsExceededReceiver;
 use Symfony\Component\Messenger\Transport\Receiver\StopWhenMessageCountIsExceededReceiver;
 use Symfony\Component\Messenger\Transport\Receiver\StopWhenTimeLimitIsReachedReceiver;
@@ -32,21 +33,25 @@
  */
 class ConsumeMessagesCommand extends Command
 {
-    protected static $defaultName = 'messenger:consume-messages';
+    protected static $defaultName = 'messenger:consume';
 
     private $busLocator;
     private $receiverLocator;
     private $logger;
     private $receiverNames;
     private $busNames;
+    private $retryStrategyLocator;
+    private $eventDispatcher;
 
-    public function __construct(ContainerInterface $busLocator, ContainerInterface $receiverLocator, LoggerInterface $logger = null, array $receiverNames = [], array $busNames = [])
+    public function __construct(ContainerInterface $busLocator, ContainerInterface $receiverLocator, LoggerInterface $logger = null, array $receiverNames = [], array $busNames = [], ContainerInterface $retryStrategyLocator = null, EventDispatcherInterface $eventDispatcher = null)
     {
         $this->busLocator = $busLocator;
         $this->receiverLocator = $receiverLocator;
         $this->logger = $logger;
         $this->receiverNames = $receiverNames;
         $this->busNames = $busNames;
+        $this->retryStrategyLocator = $retryStrategyLocator;
+        $this->eventDispatcher = $eventDispatcher;
 
         parent::__construct();
     }
@@ -132,6 +137,12 @@ protected function interact(InputInterface $input, OutputInterface $output)
      */
     protected function execute(InputInterface $input, OutputInterface $output): void
     {
+        if (false !== strpos($input->getFirstArgument(), ':consume-')) {
+            $message = 'The use of the "messenger:consume-messages" command is deprecated since version 4.3 and will be removed in 5.0. Use "messenger:consume" instead.';
+            @trigger_error($message, E_USER_DEPRECATED);
+            $output->writeln(sprintf('<comment>%s</comment>', $message));
+        }
+
         if (!$this->receiverLocator->has($receiverName = $input->getArgument('receiver'))) {
             throw new RuntimeException(sprintf('Receiver "%s" does not exist.', $receiverName));
         }
@@ -140,8 +151,13 @@ protected function execute(InputInterface $input, OutputInterface $output): void
             throw new RuntimeException(sprintf('Bus "%s" does not exist.', $busName));
         }
 
+        if (null !== $this->retryStrategyLocator && !$this->retryStrategyLocator->has($receiverName)) {
+            throw new RuntimeException(sprintf('Receiver "%s" does not have a configured retry strategy.', $receiverName));
+        }
+
         $receiver = $this->receiverLocator->get($receiverName);
         $bus = $this->busLocator->get($busName);
+        $retryStrategy = null !== $this->retryStrategyLocator ? $this->retryStrategyLocator->get($receiverName) : null;
 
         $stopsWhen = [];
         if ($limit = $input->getOption('limit')) {
@@ -174,7 +190,7 @@ protected function execute(InputInterface $input, OutputInterface $output): void
             $io->comment('Re-run the command with a -vv option to see logs about consumed messages.');
         }
 
-        $worker = new Worker($receiver, $bus);
+        $worker = new Worker($receiver, $bus, $receiverName, $retryStrategy, $this->eventDispatcher, $this->logger);
         $worker->run();
     }
 

Envelope.php

@@ -54,6 +54,18 @@ public function with(StampInterface ...$stamps): self
         return $cloned;
     }
 
+    /**
+     * @return Envelope a new Envelope instance without any stamps of the given class
+     */
+    public function withoutAll(string $stampFqcn): self
+    {
+        $cloned = clone $this;
+
+        unset($cloned->stamps[$stampFqcn]);
+
+        return $cloned;
+    }
+
     public function last(string $stampFqcn): ?StampInterface
     {
         return isset($this->stamps[$stampFqcn]) ? end($this->stamps[$stampFqcn]) : null;

Event/AbstractWorkerMessageEvent.php

@@ -0,0 +1,43 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Event;
+
+use Symfony\Component\EventDispatcher\Event;
+use Symfony\Component\Messenger\Envelope;
+
+/**
+ * @experimental in 4.3
+ */
+abstract class AbstractWorkerMessageEvent extends Event
+{
+    private $envelope;
+    private $receiverName;
+
+    public function __construct(Envelope $envelope, string $receiverName)
+    {
+        $this->envelope = $envelope;
+        $this->receiverName = $receiverName;
+    }
+
+    public function getEnvelope(): Envelope
+    {
+        return $this->envelope;
+    }
+
+    /**
+     * Returns a unique identifier for transport receiver this message was received from.
+     */
+    public function getReceiverName(): string
+    {
+        return $this->receiverName;
+    }
+}

Event/WorkerMessageFailedEvent.php

@@ -0,0 +1,45 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Event;
+
+use Symfony\Component\Messenger\Envelope;
+
+/**
+ * Dispatched when a message was received from a transport and handling failed.
+ *
+ * The event name is the class name.
+ *
+ * @experimental in 4.3
+ */
+class WorkerMessageFailedEvent extends AbstractWorkerMessageEvent
+{
+    private $throwable;
+    private $willRetry;
+
+    public function __construct(Envelope $envelope, string $receiverName, \Throwable $error, bool $willRetry)
+    {
+        $this->throwable = $error;
+        $this->willRetry = $willRetry;
+
+        parent::__construct($envelope, $receiverName);
+    }
+
+    public function getThrowable(): \Throwable
+    {
+        return $this->throwable;
+    }
+
+    public function willRetry(): bool
+    {
+        return $this->willRetry;
+    }
+}

Event/WorkerMessageHandledEvent.php

@@ -0,0 +1,23 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Event;
+
+/**
+ * Dispatched after a message was received from a transport and successfully handled.
+ *
+ * The event name is the class name.
+ *
+ * @experimental in 4.3
+ */
+class WorkerMessageHandledEvent extends AbstractWorkerMessageEvent
+{
+}

Event/WorkerMessageReceivedEvent.php

@@ -0,0 +1,23 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Event;
+
+/**
+ * Dispatched when a message was received from a transport but before sent to the bus.
+ *
+ * The event name is the class name.
+ *
+ * @experimental in 4.3
+ */
+class WorkerMessageReceivedEvent extends AbstractWorkerMessageEvent
+{
+}

Exception/MessageDecodingFailedException.php

@@ -0,0 +1,21 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Exception;
+
+/**
+ * Thrown when a message cannot be decoded in a serializer.
+ *
+ * @experimental in 4.3
+ */
+class MessageDecodingFailedException extends \InvalidArgumentException implements ExceptionInterface
+{
+}

Exception/UnrecoverableMessageHandlingException.php

@@ -0,0 +1,26 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Exception;
+
+/**
+ * Thrown while handling a message to indicate that handling will continue to fail.
+ *
+ * If something goes wrong while handling a message that's received from a transport
+ * and the message should not be retried, a handler can throw this exception.
+ *
+ * @author Frederic Bouchery <[email protected]>
+ *
+ * @experimental in 4.3
+ */
+class UnrecoverableMessageHandlingException extends RuntimeException
+{
+}

Middleware/SendMessageMiddleware.php

@@ -15,6 +15,7 @@
 use Psr\Log\NullLogger;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
 use Symfony\Component\Messenger\Stamp\SentStamp;
 use Symfony\Component\Messenger\Transport\Sender\SendersLocatorInterface;
 
@@ -54,9 +55,22 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
                 // it's a received message, do not send it back
                 $this->logger->info('Received message "{class}"', $context);
             } else {
+                /** @var RedeliveryStamp|null $redeliveryStamp */
+                $redeliveryStamp = $envelope->last(RedeliveryStamp::class);
+
                 foreach ($this->sendersLocator->getSenders($envelope, $handle) as $alias => $sender) {
+                    // on redelivery, only deliver to the given sender
+                    if (null !== $redeliveryStamp && !$redeliveryStamp->shouldRedeliverToSender(\get_class($sender), $alias)) {
+                        continue;
+                    }
+
                     $this->logger->info('Sending message "{class}" with "{sender}"', $context + ['sender' => \get_class($sender)]);
-                    $envelope = $sender->send($envelope)->with(new SentStamp(\get_class($sender), \is_string($alias) ? $alias : null));
+                    $envelope = $sender->send($envelope->with(new SentStamp(\get_class($sender), \is_string($alias) ? $alias : null)));
+                }
+
+                // on a redelivery, never call local handlers
+                if (null !== $redeliveryStamp) {
+                    $handle = false;
                 }
             }