Replace Rewrite by Around Plugin on message reject function (#15)

* - Replace Rewrite by Around Plugin on message reject function
- Fix Retries Count
- Refactoring
- Update Tests

---------

Co-authored-by: Vladyslav Sikailo <vladyslav.sikailo@run-as-root.sh>

Author: VladyslavSikailo

src/Plugin/HandleQueueMessageRejectPlugin.php

@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace RunAsRoot\MessageQueueRetry\Plugin;
+
+use JsonException;
+use Magento\Framework\MessageQueue\EnvelopeInterface;
+use Magento\Framework\MessageQueue\QueueInterface;
+use RunAsRoot\MessageQueueRetry\Exception\MessageCouldNotBeCreatedException;
+use RunAsRoot\MessageQueueRetry\Service\IsMessageShouldBeSavedForRetryService;
+use RunAsRoot\MessageQueueRetry\Service\SaveFailedMessageService;
+
+class HandleQueueMessageRejectPlugin
+{
+    public function __construct(
+        private IsMessageShouldBeSavedForRetryService $isMessageShouldBeSavedForRetryService,
+        private SaveFailedMessageService $saveFailedMessageService
+    ) {
+    }
+
+    /**
+     * @throws MessageCouldNotBeCreatedException
+     * @throws JsonException
+     */
+    public function aroundReject(
+        QueueInterface $subject,
+        callable $proceed,
+        EnvelopeInterface $envelope,
+        bool $requeue,
+        string $error
+    ): void {
+        if (!$error) {
+            $proceed($envelope, $requeue, $error);
+            return;
+        }
+
+        $shouldBeSavedForRetry = $this->isMessageShouldBeSavedForRetryService->execute($envelope);
+
+        if (!$shouldBeSavedForRetry) {
+            $proceed($envelope, $requeue, $error);
+            return;
+        }
+
+        $this->saveFailedMessageService->execute($envelope, $error);
+        $subject->acknowledge($envelope);
+    }
+}

src/Queue/Consumer.php

@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace RunAsRoot\MessageQueueRetry\Service;
+
+use Magento\Framework\MessageQueue\EnvelopeInterface;
+use PhpAmqpLib\Wire\AMQPTable;
+
+class GetMessageRetriesCountService
+{
+    public function execute(EnvelopeInterface $message): int
+    {
+        $messageProperties = $message->getProperties();
+        $applicationHeaders = $messageProperties['application_headers'] ?? null;
+
+        // If there are no application headers, then it is the first time the message has been processed.
+        if (!$applicationHeaders instanceof AMQPTable) {
+            return 0;
+        }
+
+        if (isset($applicationHeaders->getNativeData()['x-death'][0]['count'])) {
+            return $applicationHeaders->getNativeData()['x-death'][0]['count'];
+        }
+
+        return 0;
+    }
+}