Fixes yiisoft#332: Add AWS SQS FIFO support

Author: samdark

.env.example

@@ -5,4 +5,7 @@ AWS_SQS_ENABLED=
 AWS_SQS_URL=
 AWS_KEY=
 AWS_SECRET=
-AWS_REGION=
+AWS_REGION=
+AWS_SQS_FIFO_ENABLED=
+AWS_SQS_FIFO_URL=
+AWS_SQS_FIFO_MESSAGE_GROUP_ID=

CHANGELOG.md

@@ -4,6 +4,7 @@ Yii2 Queue Extension Change Log
 2.2.1 under development
 -----------------------
 
+- Enh #332: Add AWS SQS FIFO support (kringkaste, alexkart)
 - Bug #220: Updated to the latest amqp-lib (alexkart)
 
 2.2.0 Mar 20, 2019

docs/guide/driver-sqs.md

@@ -5,7 +5,7 @@ The driver uses AWS SQS to store queue data.
 
 You have to add `aws/aws-sdk-php` extension to your application in order to use it.
 
-Configuration example:
+Configuration example for standard queues:
 
 ```php
 return [
@@ -24,6 +24,30 @@ return [
 ];
 ```
 
+Configuration example for FIFO queues:
+
+```php
+return [
+    'bootstrap' => [
+        'queue', // The component registers own console commands
+    ],
+    'components' => [
+        'queue' => [
+            'class' => \yii\queue\sqs\Queue::class,
+            'url' => '<sqs url>',
+            'key' => '<key>',
+            'secret' => '<secret>',
+            'region' => '<region>',
+            'messageGroupId' => '<Group ID>',
+        ],
+    ],
+];
+```
+
+The message group ID is required by SQS for FIFO queues. You can configure your own or use the "default" value.
+
+The deduplication ID is generated automatically, so no matter if you have activated content-based deduplication in the SQS queue or not, this ID will be used.
+
 Console
 -------
 

src/drivers/sqs/Queue.php

@@ -46,6 +46,12 @@ class Queue extends CliQueue
      * @var string
      */
     public $version = 'latest';
+    /**
+     * Message Group ID for FIFO queues.
+     * @var string
+     * @since 2.2.1
+     */
+    public $messageGroupId = 'default';
     /**
      * @var string command class name
      * @inheritdoc
@@ -172,7 +178,7 @@ protected function pushMessage($message, $ttr, $delay, $priority)
             throw new NotSupportedException('Priority is not supported in this driver');
         }
 
-        $response = $this->getClient()->sendMessage([
+        $request = [
             'QueueUrl' => $this->url,
             'MessageBody' => $message,
             'DelaySeconds' => $delay,
@@ -182,7 +188,14 @@ protected function pushMessage($message, $ttr, $delay, $priority)
                     'StringValue' => $ttr,
                 ],
             ],
-        ]);
+        ];
+
+        if (substr($this->url, -5) === '.fifo') {
+            $request['MessageGroupId'] = $this->messageGroupId;
+            $request['MessageDeduplicationId'] = hash('sha256', $message);
+        }
+
+        $response = $this->getClient()->sendMessage($request);
         return $response['MessageId'];
     }
 

tests/app/config/main.php

@@ -122,4 +122,16 @@
     ];
 }
 
+if (getenv('AWS_SQS_FIFO_ENABLED')) {
+    $config['bootstrap'][] = 'sqsFifoQueue';
+    $config['components']['sqsFifoQueue'] = [
+        'class' => \yii\queue\sqs\Queue::class,
+        'url' => getenv('AWS_SQS_FIFO_URL'),
+        'key' => getenv('AWS_KEY'),
+        'secret' => getenv('AWS_SECRET'),
+        'region' => getenv('AWS_REGION'),
+        'messageGroupId' => getenv('AWS_SQS_FIFO_MESSAGE_GROUP_ID'),
+    ];
+}
+
 return $config;

tests/docker-compose.yml

@@ -34,6 +34,9 @@ services:
       AWS_SECRET: ${AWS_SECRET}
       AWS_REGION: ${AWS_REGION}
       AWS_SQS_URL: ${AWS_SQS_URL}
+      AWS_SQS_FIFO_ENABLED: ${AWS_SQS_FIFO_ENABLED}
+      AWS_SQS_FIFO_URL: ${AWS_SQS_FIFO_URL}
+      AWS_SQS_FIFO_MESSAGE_GROUP_ID: ${AWS_SQS_FIFO_MESSAGE_GROUP_ID}
     depends_on: &php_depends_on
     - mysql
     - postgres

tests/drivers/CliTestCase.php

@@ -62,7 +62,7 @@ private function prepareCmd($cmd)
 
         return strtr($cmd, [
             'php' => PHP_BINARY,
-            'yii' => 'tests/yii',
+            'yii' => __DIR__ . '/../yii',
             'queue' => $method->invoke($this->getQueue()),
         ]);
     }

tests/drivers/sqs/FifoQueueTest.php

@@ -0,0 +1,84 @@
+<?php
+/**
+ * @link http://www.yiiframework.com/
+ * @copyright Copyright (c) 2008 Yii Software LLC
+ * @license http://www.yiiframework.com/license/
+ */
+
+namespace tests\drivers\sqs;
+
+use tests\app\RetryJob;
+use tests\drivers\CliTestCase;
+use Yii;
+use yii\queue\sqs\Queue;
+
+/**
+ * SQS FIFO Queue Test.
+ */
+class FifoQueueTest extends CliTestCase
+{
+    public function testRun()
+    {
+        $job = $this->createSimpleJob();
+        $this->getQueue()->push($job);
+        $this->runProcess('php yii queue/run');
+
+        $this->assertSimpleJobDone($job);
+    }
+
+    public function testListen()
+    {
+        $this->startProcess('php yii queue/listen 1');
+        $job = $this->createSimpleJob();
+        $this->getQueue()->push($job);
+
+        $this->assertSimpleJobDone($job);
+    }
+
+    public function testFifoQueueDoesNotSupportPerMessageDelays()
+    {
+        $this->startProcess('php yii queue/listen 1');
+        $job = $this->createSimpleJob();
+
+        $this->setExpectedException('\Aws\Sqs\Exception\SqsException');
+        $this->getQueue()->delay(2)->push($job);
+    }
+
+    public function testRetry()
+    {
+        $this->startProcess('php yii queue/listen 1');
+        $job = new RetryJob(['uid' => uniqid()]);
+        $this->getQueue()->push($job);
+        sleep(6);
+
+        $this->assertFileExists($job->getFileName());
+        $this->assertEquals('aa', file_get_contents($job->getFileName()));
+    }
+
+    public function testClear()
+    {
+        if (!getenv('AWS_SQS_FIFO_CLEAR_TEST_ENABLED')) {
+            $this->markTestSkipped(__METHOD__ . ' is disabled');
+        }
+
+        $this->getQueue()->push($this->createSimpleJob());
+        $this->runProcess('php yii queue/clear --interactive=0');
+    }
+
+    /**
+     * @return Queue
+     */
+    protected function getQueue()
+    {
+        return Yii::$app->sqsFifoQueue;
+    }
+
+    protected function setUp()
+    {
+        if (!getenv('AWS_SQS_FIFO_ENABLED')) {
+            $this->markTestSkipped('AWS SQS FIFO tests are disabled');
+        }
+
+        parent::setUp();
+    }
+}