Merge pull request #672 from php-enqueue/sns

Amazon SNS transport. Publish-Subscribe support based on combination of SQS and SNQ

Author: makasim

bin/subtree-split

@@ -58,6 +58,8 @@ remote rdkafka [email protected]:php-enqueue/rdkafka.git
 remote dbal [email protected]:php-enqueue/dbal.git
 remote null [email protected]:php-enqueue/null.git
 remote sqs [email protected]:php-enqueue/sqs.git
+remote sns [email protected]:php-enqueue/sns.git
+remote snsqs [email protected]:php-enqueue/snsqs.git
 remote gps [email protected]:php-enqueue/gps.git
 remote enqueue-bundle [email protected]:php-enqueue/enqueue-bundle.git
 remote job-queue [email protected]:php-enqueue/job-queue.git
@@ -84,6 +86,8 @@ split 'pkg/redis' redis
 split 'pkg/dbal' dbal
 split 'pkg/null' null
 split 'pkg/sqs' sqs
+split 'pkg/sns' sns
+split 'pkg/snsqs' snsqs
 split 'pkg/gps' gps
 split 'pkg/enqueue-bundle' enqueue-bundle
 split 'pkg/job-queue' job-queue

composer.json

@@ -25,7 +25,7 @@
     "doctrine/orm": "~2.4",
     "mongodb/mongodb": "^1.2",
     "pda/pheanstalk": "^3",
-    "aws/aws-sdk-php": "~3.26",
+    "aws/aws-sdk-php": "^3.26",
     "stomp-php/stomp-php": "^4",
     "php-http/guzzle6-adapter": "^1.1",
     "php-http/client-common": "^1.7@dev",
@@ -76,6 +76,8 @@
       "Enqueue\\Redis\\": "pkg/redis/",
       "Enqueue\\SimpleClient\\": "pkg/simple-client/",
       "Enqueue\\Sqs\\": "pkg/sqs/",
+      "Enqueue\\Sns\\": "pkg/sns/",
+      "Enqueue\\SnsQs\\": "pkg/snsqs/",
       "Enqueue\\Stomp\\": "pkg/stomp/",
       "Enqueue\\Test\\": "pkg/test/",
       "Enqueue\\Dsn\\": "pkg/dsn/",

docker-compose.yml

@@ -34,6 +34,8 @@ services:
       - PHPREDIS_DSN=redis+phpredis://redis
       - GPS_DSN=gps:?projectId=mqdev&emulatorHost=http://google-pubsub:8085
       - SQS_DSN=sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4576&version=latest
+      - SNS_DSN=sns:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4575&version=latest
+      - SNSQS_DSN=snsqs:?key=key&secret=secret&region=us-east-1&sns_endpoint=http://localstack:4575&sqs_endpoint=http://localstack:4576&version=latest
       - WAMP_DSN=wamp://thruway:9090
       - REDIS_HOST=redis
       - REDIS_PORT=6379
@@ -129,9 +131,10 @@ services:
     image: 'localstack/localstack:latest'
     ports:
       - '4576:4576'
+      - '4575:4575'
     environment:
       HOSTNAME_EXTERNAL: 'localstack'
-      SERVICES: 'sqs'
+      SERVICES: 'sqs,sns'
 
   influxdb:
     image: 'influxdb:latest'

docs/index.md

@@ -12,6 +12,7 @@ Enqueue is an MIT-licensed open source project with its ongoing development made
 * [Quick tour](quick_tour.md)
 * [Transports](#transports)
     - Amqp based on [the ext](transport/amqp.md), [bunny](transport/amqp_bunny.md), [the lib](transport/amqp_lib.md)
+    - [Amazon SNS-SQS](transport/snsqs.md)
     - [Amazon SQS](transport/sqs.md)
     - [Google PubSub](transport/gps.md)
     - [Beanstalk (Pheanstalk)](transport/pheanstalk.md)

docs/transport/snsqs.md

@@ -0,0 +1,182 @@
+ <h2 align="center">Supporting Enqueue</h2>
+
+Enqueue is an MIT-licensed open source project with its ongoing development made possible entirely by the support of community and our customers. If you'd like to join them, please consider:
+
+- [Become a sponsor](https://www.patreon.com/makasim)
+- [Become our client](http://forma-pro.com/)
+
+---
+
+# Amazon SNS-SQS transport
+
+Utilize two Amazon services [SNS-SQS](https://docs.aws.amazon.com/sns/latest/dg/sns-sqs-as-subscriber.html) to
+implement [Publish-Subscribe](https://www.enterpriseintegrationpatterns.com/patterns/messaging/PublishSubscribeChannel.html)
+enterprise integration pattern. As opposed to single SQS transport this adds ability to use [MessageBus](https://www.enterpriseintegrationpatterns.com/patterns/messaging/MessageBus.html)
+with enqueue.
+
+A transport for [Amazon SQS](https://aws.amazon.com/sqs/) broker.
+It uses internally official [aws sdk library](https://packagist.org/packages/aws/aws-sdk-php) 
+
+* [Installation](#installation)
+* [Create context](#create-context)
+* [Declare topic, queue and bind them together](#declare-topic-queue-and-bind-them-together)
+* [Send message to topic](#send-message-to-topic)
+* [Send message to queue](#send-message-to-queue)
+* [Consume message](#consume-message)
+* [Purge queue messages](#purge-queue-messages)
+* [Queue from another AWS account](#queue-from-another-aws-account)
+
+## Installation
+
+```bash
+$ composer require enqueue/sqs
+```
+
+## Create context
+
+```php
+<?php
+use Enqueue\SnsQs\SnsQsConnectionFactory;
+ 
+$factory = new SnsQsConnectionFactory([
+    'key' => 'aKey',              
+    'secret' => 'aSecret',       
+    'region' => 'aRegion',
+    
+    // or you can segregate options using prefixes "sns_", "sqs_"
+    'key' => 'aKey',              // common option for both SNS and SQS
+    'sns_region' => 'aSnsRegion', // SNS transport option
+    'sqs_region' => 'aSqsRegion', // SQS transport option
+]);
+
+// same as above but given as DSN string. You may need to url encode secret if it contains special char (like +)
+$factory = new SnsQsConnectionFactory('snsqs:?key=aKey&secret=aSecret&region=aRegion');
+
+$context = $factory->createContext();
+
+// if you have enqueue/enqueue library installed you can use a factory to build context from DSN 
+$context = (new \Enqueue\ConnectionFactoryFactory())->create('snsqs:')->createContext();
+```
+
+## Declare topic, queue and bind them together
+
+Declare topic, queue operation creates a topic, queue on a broker side.
+Bind creates connection between topic and queue. You publish message to
+the topic and topic sends message to each queue connected to the topic.
+
+ 
+```php
+<?php
+/** @var \Enqueue\SnsQs\SnsQsContext $context */
+
+$inTopic = $context->createTopic('in');
+$context->declareTopic($inTopic);
+
+$out1Queue = $context->createQueue('out1');
+$context->declareQueue($out1Queue);
+
+$out2Queue = $context->createQueue('out2');
+$context->declareQueue($out2Queue);
+
+$context->bind($inTopic, $out1Queue);
+$context->bind($inTopic, $out2Queue);
+
+// to remove topic/queue use deleteTopic/deleteQueue method
+//$context->deleteTopic($inTopic);
+//$context->deleteQueue($out1Queue);
+//$context->unbind(inTopic, $out1Queue);
+```
+
+## Send message to topic 
+
+```php
+<?php
+/** @var \Enqueue\SnsQs\SnsQsContext $context */
+
+$inTopic = $context->createTopic('in');
+$message = $context->createMessage('Hello world!');
+
+$context->createProducer()->send($inTopic, $message);
+```
+
+## Send message to queue 
+
+You can bypass topic and publish message directly to the queue
+
+```php
+<?php
+/** @var \Enqueue\SnsQs\SnsQsContext $context */
+
+$fooQueue = $context->createQueue('foo');
+$message = $context->createMessage('Hello world!');
+
+$context->createProducer()->send($fooQueue, $message);
+```
+
+
+## Consume message:
+
+```php
+<?php
+/** @var \Enqueue\SnsQs\SnsQsContext $context */
+
+$out1Queue = $context->createQueue('out1');
+$consumer = $context->createConsumer($out1Queue);
+
+$message = $consumer->receive();
+
+// process a message
+
+$consumer->acknowledge($message);
+// $consumer->reject($message);
+```
+
+## Purge queue messages:
+
+```php
+<?php
+/** @var \Enqueue\SnsQs\SnsQsContext $context */
+
+$fooQueue = $context->createQueue('foo');
+
+$context->purgeQueue($fooQueue);
+```
+
+## Queue from another AWS account
+
+SQS allows to use queues from another account. You could set it globally for all queues via option `queue_owner_aws_account_id` or 
+per queue using `SnsQsQueue::setQueueOwnerAWSAccountId` method.
+
+```php
+<?php
+use Enqueue\SnsQs\SnsQsConnectionFactory;
+
+// globally for all queues
+$factory = new SnsQsConnectionFactory('snsqs:?sqs_queue_owner_aws_account_id=awsAccountId');
+
+$context = (new SnsQsConnectionFactory('snsqs:'))->createContext();
+
+// per queue.
+$queue = $context->createQueue('foo');
+$queue->setQueueOwnerAWSAccountId('awsAccountId');
+```
+
+## Multi region examples
+
+Enqueue SNSQS provides a generic multi-region support. This enables users to specify which AWS Region to send a command to by setting region on SnsQsQueue.
+If not specified the default region is used. 
+
+```php
+<?php
+use Enqueue\SnsQs\SnsQsConnectionFactory;
+
+$context = (new SnsQsConnectionFactory('snsqs:?region=eu-west-2'))->createContext();
+
+$queue = $context->createQueue('foo');
+$queue->setRegion('us-west-2');
+
+// the request goes to US West (Oregon) Region
+$context->declareQueue($queue);
+```
+
+[back to index](../index.md)

phpunit.xml.dist

@@ -61,6 +61,14 @@
             <directory>pkg/sqs/Tests</directory>
         </testsuite>
 
+        <testsuite name="sns transport">
+            <directory>pkg/sns/Tests</directory>
+        </testsuite>
+
+        <testsuite name="snsqs transport">
+            <directory>pkg/snsqs/Tests</directory>
+        </testsuite>
+
         <testsuite name="pheanstalk transport">
             <directory>pkg/pheanstalk/Tests</directory>
         </testsuite>

pkg/enqueue-bundle/Tests/Functional/UseCasesTest.php

@@ -143,6 +143,14 @@ public function provideEnqueueConfigs()
             ],
         ]];
 
+        yield 'snsqs' => [[
+            'default' => [
+                'transport' => [
+                    'dsn' => getenv('SNSQS_DSN'),
+                ],
+            ],
+        ]];
+
 //
 //        yield 'gps' => [[
 //            'transport' => [

pkg/enqueue/Client/Driver/SnsQsDriver.php

@@ -0,0 +1,90 @@
+<?php
+
+namespace Enqueue\Client\Driver;
+
+use Enqueue\SnsQs\SnsQsContext;
+use Enqueue\SnsQs\SnsQsTopic;
+use Interop\Queue\Destination;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+
+/**
+ * @method SnsQsContext getContext()
+ * @method SnsQsTopic   createRouterTopic()
+ */
+class SnsQsDriver extends GenericDriver
+{
+    public function __construct(SnsQsContext $context, ...$args)
+    {
+        parent::__construct($context, ...$args);
+    }
+
+    public function setupBroker(LoggerInterface $logger = null): void
+    {
+        $logger = $logger ?: new NullLogger();
+        $log = function ($text, ...$args) use ($logger) {
+            $logger->debug(sprintf('[SqsQsDriver] '.$text, ...$args));
+        };
+
+        // setup router
+        $routerTopic = $this->createRouterTopic();
+        $log('Declare router topic: %s', $routerTopic->getTopicName());
+        $this->getContext()->declareTopic($routerTopic);
+
+        $routerQueue = $this->createQueue($this->getConfig()->getRouterQueue());
+        $log('Declare router queue: %s', $routerQueue->getQueueName());
+        $this->getContext()->declareQueue($routerQueue);
+
+        $log('Bind router queue to topic: %s -> %s', $routerQueue->getQueueName(), $routerTopic->getTopicName());
+        $this->getContext()->bind($routerTopic, $routerQueue);
+
+        // setup queues
+        $declaredQueues = [];
+        $declaredTopics = [];
+        foreach ($this->getRouteCollection()->all() as $route) {
+            $queue = $this->createRouteQueue($route);
+            if (false === array_key_exists($queue->getQueueName(), $declaredQueues)) {
+                $log('Declare processor queue: %s', $queue->getQueueName());
+                $this->getContext()->declareQueue($queue);
+
+                $declaredQueues[$queue->getQueueName()] = true;
+            }
+
+            if ($route->isCommand()) {
+                continue;
+            }
+
+            $topic = $this->doCreateTopic($this->createTransportQueueName($route->getSource(), true));
+            if (false === array_key_exists($topic->getTopicName(), $declaredTopics)) {
+                $log('Declare processor topic: %s', $topic->getTopicName());
+                $this->getContext()->declareTopic($topic);
+
+                $declaredTopics[$topic->getTopicName()] = true;
+            }
+
+            $log('Bind processor queue to topic: %s -> %s', $queue->getQueueName(), $topic->getTopicName());
+            $this->getContext()->bind($topic, $queue);
+        }
+    }
+
+    protected function createRouterTopic(): Destination
+    {
+        return $this->doCreateTopic(
+            $this->createTransportRouterTopicName($this->getConfig()->getRouterTopic(), true)
+        );
+    }
+
+    protected function createTransportRouterTopicName(string $name, bool $prefix): string
+    {
+        $name = parent::createTransportRouterTopicName($name, $prefix);
+
+        return str_replace('.', '_dot_', $name);
+    }
+
+    protected function createTransportQueueName(string $name, bool $prefix): string
+    {
+        $name = parent::createTransportQueueName($name, $prefix);
+
+        return str_replace('.', '_dot_', $name);
+    }
+}

pkg/enqueue/Client/Resources.php

@@ -12,6 +12,7 @@
 use Enqueue\Client\Driver\RabbitMqStompDriver;
 use Enqueue\Client\Driver\RdKafkaDriver;
 use Enqueue\Client\Driver\RedisDriver;
+use Enqueue\Client\Driver\SnsQsDriver;
 use Enqueue\Client\Driver\SqsDriver;
 use Enqueue\Client\Driver\StompDriver;
 
@@ -92,6 +93,12 @@ public static function getKnownDrivers(): array
                 'requiredSchemeExtensions' => [],
                 'packages' => ['enqueue/enqueue', 'enqueue/sqs'],
             ];
+            $map[] = [
+                'schemes' => ['snsqs'],
+                'driverClass' => SnsQsDriver::class,
+                'requiredSchemeExtensions' => [],
+                'packages' => ['enqueue/enqueue', 'enqueue/sqs', 'enqueue/sns', 'enqueue/snsqs'],
+            ];
             $map[] = [
                 'schemes' => ['stomp'],
                 'driverClass' => StompDriver::class,