feat(batching): add support for message batching to AMQP adapter

Author: basert

src/Queue/Broker/AMQP.php

@@ -14,6 +14,8 @@
 use Utopia\Queue\Message;
 use Utopia\Queue\Publisher;
 use Utopia\Queue\Queue;
+use Utopia\Queue\Result\Commit;
+use Utopia\Queue\Result\NoCommit;
 
 class AMQP implements Publisher, Consumer
 {
@@ -81,8 +83,12 @@ public function consume(Queue $queue, callable $messageCallback, callable $succe
                 $nextMessage['timestamp'] = (int)$nextMessage['timestamp'];
                 $message = new Message($nextMessage);
 
-                $messageCallback($message);
-                $amqpMessage->ack();
+                $result = $messageCallback($message);
+                match(true) {
+                    $result instanceof Commit => $amqpMessage->ack(true),
+                    $result instanceof NoCommit => null,
+                    default => $amqpMessage->ack()
+                };
                 $successCallback($message);
             } catch (Retryable $e) {
                 $amqpMessage->nack(requeue: true);

src/Queue/Consumer.php

@@ -2,11 +2,14 @@
 
 namespace Utopia\Queue;
 
+use Utopia\Queue\Result\Commit;
+use Utopia\Queue\Result\NoCommit;
+
 interface Consumer
 {
     /**
      * @param Queue $queue
-     * @param callable(Message $message): void $messageCallback
+     * @param callable(Message $message): Commit|NoCommit|mixed $messageCallback
      * @param callable(Message $message): void $successCallback
      * @param callable(Message $message, \Throwable $th): void $errorCallback
      * @return void

src/Queue/Result/Commit.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace Utopia\Queue\Result;
+
+class Commit
+{
+}

src/Queue/Result/NoCommit.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace Utopia\Queue\Result;
+
+class NoCommit
+{
+}

src/Queue/Server.php

@@ -249,7 +249,7 @@ function (Message $message) {
                                     }
                                 }
 
-                                \call_user_func_array($this->job->getAction(), $this->getArguments($this->job, $message->getPayload()));
+                                return \call_user_func_array($this->job->getAction(), $this->getArguments($this->job, $message->getPayload()));
                             } finally {
                                 $processDuration = microtime(true) - $receivedAtTimestamp;
                                 $this->processDuration->record($processDuration);