From 1f2a80f820f27d47f71038450c7573487704371a Mon Sep 17 00:00:00 2001
From: Konstantin Grachev <me@grachevko.ru>
Date: Mon, 28 Jun 2021 11:28:39 +0300
Subject: [PATCH] Non retry messages must be acknowledged

---
 .../AckUnrecoverableMessageListener.php       | 37 +++++++++++++++++++
 src/Resources/config/services.yml             |  4 ++
 2 files changed, 41 insertions(+)
 create mode 100644 src/Messenger/AckUnrecoverableMessageListener.php

diff --git a/src/Messenger/AckUnrecoverableMessageListener.php b/src/Messenger/AckUnrecoverableMessageListener.php
new file mode 100644
index 0000000..eee6ca7
--- /dev/null
+++ b/src/Messenger/AckUnrecoverableMessageListener.php
@@ -0,0 +1,37 @@
+<?php
+declare(strict_types=1);
+
+namespace Nsq\NsqBundle\Messenger;
+
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use function Amp\Promise\wait;
+
+final class AckUnrecoverableMessageListener implements EventSubscriberInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            WorkerMessageFailedEvent::class => ['onMessageFailed', 500],
+        ];
+    }
+
+    public function onMessageFailed(WorkerMessageFailedEvent $event): void
+    {
+        if ($event->willRetry()) {
+            return;
+        }
+
+        $envelope = $event->getEnvelope();
+        $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
+
+        if ($message->isProcessed()) {
+            return;
+        }
+
+        wait($message->finish());
+    }
+}
diff --git a/src/Resources/config/services.yml b/src/Resources/config/services.yml
index c0792c7..85f8187 100644
--- a/src/Resources/config/services.yml
+++ b/src/Resources/config/services.yml
@@ -3,3 +3,7 @@ services:
         tags:
             - 'messenger.transport_factory'
             - 'container.no_preload'
+
+    Nsq\NsqBundle\Messenger\AckUnrecoverableMessageListener:
+        tags:
+            - { name: kernel.event_subscriber }