[Messenger] Add a way to no ack message automatically

lyrixx · lyrixx · commit 89bb00850ecb · 2021-09-03T12:46:19.000+02:00
diff --git a/src/Symfony/Component/Messenger/Handler/ConfigureAutoAckInterface.php b/src/Symfony/Component/Messenger/Handler/ConfigureAutoAckInterface.php
@@ -0,0 +1,24 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Handler;
+
+use Symfony\Component\Messenger\Envelope;
+
+/**
+ * Marker interface for message handlers to configure if auto ACK is disabled.
+ *
+ * @author Grégoire Pineau <lyrixx@lyrixx.info>
+ */
+interface ConfigureAutoAckInterface
+{
+    public function disableAutoAck(Envelope $envelope): bool;
+}
diff --git a/src/Symfony/Component/Messenger/Middleware/HandleMessageMiddleware.php b/src/Symfony/Component/Messenger/Middleware/HandleMessageMiddleware.php
@@ -16,8 +16,10 @@
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\HandlerFailedException;
 use Symfony\Component\Messenger\Exception\NoHandlerForMessageException;
+use Symfony\Component\Messenger\Handler\ConfigureAutoAckInterface;
 use Symfony\Component\Messenger\Handler\HandlerDescriptor;
 use Symfony\Component\Messenger\Handler\HandlersLocatorInterface;
+use Symfony\Component\Messenger\Stamp\DelayedAckStamp;
 use Symfony\Component\Messenger\Stamp\HandledStamp;
 
 /**
@@ -60,7 +62,10 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
 
             try {
                 $handler = $handlerDescriptor->getHandler();
-                $handledStamp = HandledStamp::fromDescriptor($handlerDescriptor, $handler($message));
+                if ($handler instanceof ConfigureAutoAckInterface && $handler->disableAutoAck($envelope)) {
+                    $envelope = $envelope->with(new DelayedAckStamp());
+                }
+                $handledStamp = HandledStamp::fromDescriptor($handlerDescriptor, $handler($message, $envelope));
                 $envelope = $envelope->with($handledStamp);
                 $this->logger->info('Message {class} handled by {handler}', $context + ['handler' => $handledStamp->getHandlerName()]);
             } catch (\Throwable $e) {
diff --git a/src/Symfony/Component/Messenger/Stamp/DelayedAckStamp.php b/src/Symfony/Component/Messenger/Stamp/DelayedAckStamp.php
@@ -0,0 +1,19 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Stamp;
+
+/**
+ * Apply this stamp to delay ACK of your message on a transport.
+ */
+final class DelayedAckStamp implements StampInterface
+{
+}
diff --git a/src/Symfony/Component/Messenger/Worker.php b/src/Symfony/Component/Messenger/Worker.php
@@ -24,6 +24,7 @@
 use Symfony\Component\Messenger\Exception\RejectRedeliveredMessageException;
 use Symfony\Component\Messenger\Exception\RuntimeException;
 use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
+use Symfony\Component\Messenger\Stamp\DelayedAckStamp;
 use Symfony\Component\Messenger\Stamp\ReceivedStamp;
 use Symfony\Component\Messenger\Transport\Receiver\QueueReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
@@ -165,7 +166,9 @@ private function handleMessage(Envelope $envelope, ReceiverInterface $receiver,
             $this->logger->info('{class} was handled successfully (acknowledging to transport).', $context);
         }
 
-        $receiver->ack($envelope);
+        if (null === $envelope->last(DelayedAckStamp::class)) {
+            $receiver->ack($envelope);
+        }
     }
 
     public function stop(): void
