SWR-15367

Author: NekloShcherbyna

src/Services/QueueService.php

@@ -6,6 +6,7 @@
 use Psr\Log\LoggerInterface;
 use Salesmessage\LibRabbitMQ\Dto\VhostApiDto;
 use Salesmessage\LibRabbitMQ\Services\Api\RabbitApiClient;
+use Throwable;
 
 class QueueService
 {

src/VhostsConsumer.php

@@ -365,15 +365,14 @@ private function processBatch(RabbitMQQueue $connection): void
             }
 
             $this->connectionMutex->lock(static::MAIN_HANDLER_LOCK);
-            foreach ($batchJobMessages as $batchMessage) {
-                if ($isBatchSuccess) {
-                    $this->ackMessage($batchMessage);
-
-                    continue;
+            if ($isBatchSuccess) {
+                $lastBatchMessage = end($batchJobMessages);
+                $this->ackMessage($lastBatchMessage, true);
+            } else {
+                foreach ($batchJobMessages as $batchMessage) {
+                    $job = $this->getJobByMessage($batchMessage, $connection);
+                    $this->processSingleJob($job);
                 }
-
-                $job = $this->getJobByMessage($batchMessage, $connection);
-                $this->processSingleJob($job, $connection);
             }
             $this->connectionMutex->unlock(static::MAIN_HANDLER_LOCK);
         }
@@ -423,12 +422,13 @@ private function processSingleJob(RabbitMQJob $job): void
 
     /**
      * @param AMQPMessage $message
+     * @param bool $multiple
      * @return void
      */
-    private function ackMessage(AMQPMessage $message): void
+    private function ackMessage(AMQPMessage $message, bool $multiple = false): void
     {
         try {
-            $message->ack();
+            $message->ack($multiple);
         } catch (Throwable $exception) {
             $this->output->error('Ack message error: ' . $exception->getMessage());
         }