Merge pull request #16 from plastiik/TES-3412-Add-fifo-support-to-queues

TES-3412 Add support for FIFO queues.
SykesCottages · Mar 26, 2021 · 7105304 · 7105304
2 parents d33c427 + 39dd901
commit 7105304
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 7 deletions.
diff --git a/src/Connector/Contract/Queue.php b/src/Connector/Contract/Queue.php
@@ -11,7 +11,12 @@ interface Queue
     /**
      * @param string[] $message
      */
-    public function queueMessage(string $queue, array $message) : void;
+    public function queueMessage(
+        string $queue,
+        array $message,
+        ?string $messageId = null,
+        ?string $duplicationId = null
+    ) : void;
 
     public function consume(string $queue, callable $callback, callable $idleCallback) : void;
 

diff --git a/src/Connector/RabbitMQ.php b/src/Connector/RabbitMQ.php
@@ -37,8 +37,12 @@ class RabbitMQ extends AMQPLazyConnection implements Queue
     /**
      * @param string[] $message
      */
-    public function queueMessage(string $queue, array $message) : void
-    {
+    public function queueMessage(
+        string $queue,
+        array $message,
+        ?string $messageId = null,
+        ?string $duplicationId = null
+    ) : void {
         $this->connectToChannel();
 
         $this->channel->basic_publish(

diff --git a/src/Connector/SQS.php b/src/Connector/SQS.php
@@ -29,13 +29,27 @@ class SQS extends SqsClient implements Queue
     /**
      * @param string[] $message
      */
-    public function queueMessage(string $queue, array $message) : void
-    {
-        $this->sendMessage([
+    public function queueMessage(
+        string $queue,
+        array $message,
+        ?string $messageId = null,
+        ?string $duplicationId = null
+    ) : void {
+        $message = [
             'QueueUrl' => $queue,
             'MessageBody' => json_encode($message),
             'MessageAttributes' => $this->getMessageAttributes($message),
-        ]);
+        ];
+
+        if ($messageId) {
+            $message['MessageGroupId'] = $messageId;
+        }
+
+        if ($duplicationId) {
+            $message['MessageDeduplicationId'] = $duplicationId;
+        }
+
+        $this->sendMessage($message);
     }
 
     public function consume(string $queue, callable $callback, callable $idleCallback) : void