Amazon SQS Transport

README.md

@@ -9,7 +9,14 @@ Features:
 
 * [JMS](https://docs.oracle.com/javaee/7/api/javax/jms/package-summary.html) like transport [abstraction](https://github.com/php-enqueue/psr-queue).
 * Feature rich.
-* Supports [AMQP](docs/transport/amqp.md) (RabbitMQ, ActiveMQ and others), [STOMP](docs/transport/stomp.md) (RabbitMQ, ActiveMQ and others), [Redis](docs/transport/redis.md), [Doctrine DBAL](docs/transport/dbal.md), [Filesystem](docs/transport/filesystem.md), [Null](docs/transport/null.md) transports.
+* Supports  transports:
+    - [AMQP](docs/transport/amqp.md) (RabbitMQ, ActiveMQ and others), 
+    - [STOMP](docs/transport/stomp.md)
+    - [Amazon SQS](docs/transport/sqs.md)
+    - [Redis](docs/transport/redis.md)
+    - [Doctrine DBAL](docs/transport/dbal.md)
+    - [Filesystem](docs/transport/filesystem.md)
+    - [Null](docs/transport/null.md).
 * Generic purpose abstraction level (the transport level).
 * "Opinionated" easy to use abstraction level (the client level).
 * [Message bus](http://www.enterpriseintegrationpatterns.com/patterns/messaging/MessageBus.html) support.

bin/dev

@@ -3,7 +3,7 @@
 set -x
 set -e
 
-while getopts "bustefc" OPTION; do
+while getopts "bustefcd" OPTION; do
   case $OPTION in
     b)
       COMPOSE_PROJECT_NAME=mqdev docker-compose build
@@ -27,6 +27,8 @@ while getopts "bustefc" OPTION; do
       COMPOSE_PROJECT_NAME=mqdev docker-compose run -e CHANGELOG_GITHUB_TOKEN=${CHANGELOG_GITHUB_TOKEN:-""} --workdir="/mqdev" --rm generate-changelog github_changelog_generator --future-release "$2" --simple-list
       ;;
 
+    d) COMPOSE_PROJECT_NAME=mqdev docker-compose run --workdir="/mqdev" --rm dev php pkg/enqueue-bundle/Tests/Functional/app/console.php config:dump-reference enqueue
+      ;;
     \?)
       echo "Invalid option: -$OPTARG" >&2
       exit 1

bin/subtree-split

@@ -51,6 +51,7 @@ remote fs git@github.com:php-enqueue/fs.git
 remote redis git@github.com:php-enqueue/redis.git
 remote dbal git@github.com:php-enqueue/dbal.git
 remote null git@github.com:php-enqueue/null.git
+remote sqs git@github.com:php-enqueue/sqs.git
 remote enqueue-bundle git@github.com:php-enqueue/enqueue-bundle.git
 remote job-queue git@github.com:php-enqueue/job-queue.git
 remote test git@github.com:php-enqueue/test.git
@@ -63,6 +64,7 @@ split 'pkg/fs' fs
 split 'pkg/redis' redis
 split 'pkg/dbal' dbal
 split 'pkg/null' null
+split 'pkg/sqs' sqs
 split 'pkg/enqueue-bundle' enqueue-bundle
 split 'pkg/job-queue' job-queue
 split 'pkg/test' test

composer.json

@@ -12,6 +12,7 @@
     "enqueue/fs": "*@dev",
     "enqueue/null": "*@dev",
     "enqueue/dbal": "*@dev",
+    "enqueue/sqs": "*@dev",
     "enqueue/enqueue-bundle": "*@dev",
     "enqueue/job-queue": "*@dev",
     "enqueue/test": "*@dev",
@@ -72,6 +73,10 @@
     {
       "type": "path",
       "url": "pkg/dbal"
+    },
+    {
+      "type": "path",
+      "url": "pkg/sqs"
     }
   ]
 }

docker-compose.yml

@@ -24,6 +24,9 @@ services:
       - SYMFONY__DB__PASSWORD=rootpass
       - SYMFONY__REDIS__HOST=redis
       - SYMFONY__REDIS__PORT=6379
+      - AWS__SQS__KEY=$ENQUEUE_AWS__SQS__KEY
+      - AWS__SQS__SECRET=$ENQUEUE_AWS__SQS__SECRET
+      - AWS__SQS__REGION=$ENQUEUE_AWS__SQS__REGION
 
   rabbitmq:
     image: enqueue/rabbitmq:latest

docs/bundle/config_reference.md

@@ -130,6 +130,16 @@ enqueue:
             # How often query for new messages.
             polling_interval:     1000
             lazy:                 true
+        sqs:
+            key:                  null
+            secret:               null
+            token:                null
+            region:               ~ # Required
+            retries:              3
+            version:              '2012-11-05'
+
+            # the connection will be performed as later as possible, if the option set to true
+            lazy:                 true
     client:
         traceable_producer:   false
         prefix:               enqueue

docs/index.md

@@ -3,7 +3,8 @@
 * [Quick tour](quick_tour.md)
 * Transports
     - [Amqp (RabbitMQ, ActiveMQ)](transport/amqp.md)
-    - [Stomp (RabbitMQ, ActiveMQ)](transport/stomp.md)
+    - [Amazon SQS](transport/sqs.md)
+    - [Stomp](transport/stomp.md)
     - [Redis](transport/redis.md)
     - [Doctrine DBAL](transport/dbal.md)
     - [Filesystem](transport/filesystem.md)

docs/transport/sqs.md

@@ -0,0 +1,89 @@
+# Amazon SQS transport
+
+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 queue](#decalre-queue)
+* [Send message to queue](#send-message-to-queue)
+* [Consume message](#consume-message)
+* [Purge queue messages](#purge-queue-messages)
+
+## Installation
+
+```bash
+$ composer require enqueue/sqs
+```
+
+## Create context
+
+```php
+<?php
+use Enqueue\Sqs\SqsConnectionFactory;
+
+$connectionFactory = new SqsConnectionFactory([
+    'key' => 'aKey',
+    'secret' => 'aSecret',
+    'region' => 'aRegion',
+]);
+
+$psrContext = $connectionFactory->createContext();
+```
+
+## Declare queue.
+
+Declare queue operation creates a queue on a broker side. 
+
+```php
+<?php
+/** @var \Enqueue\Sqs\SqsContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('foo');
+$psrContext->declareQueue($fooQueue);
+
+// to remove queue use deleteQueue method
+//$psrContext->deleteQueue($fooQueue);
+```
+
+## Send message to queue 
+
+```php
+<?php
+/** @var \Enqueue\Sqs\SqsContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('foo');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->createProducer()->send($fooQueue, $message);
+```
+
+## Consume message:
+
+```php
+<?php
+/** @var \Enqueue\Sqs\SqsContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('foo');
+$consumer = $psrContext->createConsumer($fooQueue);
+
+$message = $consumer->receive();
+
+// process a message
+
+$consumer->acknowledge($message);
+// $consumer->reject($message);
+```
+
+## Purge queue messages:
+
+```php
+<?php
+/** @var \Enqueue\Sqs\SqsContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('foo');
+
+$psrContext->purge($fooQueue);
+```
+
+[back to index](../index.md)

phpunit.xml.dist

@@ -45,6 +45,10 @@
             <directory>pkg/null/Tests</directory>
         </testsuite>
 
+        <testsuite name="sqs transport">
+            <directory>pkg/sqs/Tests</directory>
+        </testsuite>
+
         <testsuite name="enqueue-bundle">
             <directory>pkg/enqueue-bundle/Tests</directory>
         </testsuite>

pkg/enqueue-bundle/EnqueueBundle.php

@@ -17,6 +17,8 @@
 use Enqueue\Fs\Symfony\FsTransportFactory;
 use Enqueue\Redis\RedisContext;
 use Enqueue\Redis\Symfony\RedisTransportFactory;
+use Enqueue\Sqs\SqsContext;
+use Enqueue\Sqs\Symfony\SqsTransportFactory;
 use Enqueue\Stomp\StompContext;
 use Enqueue\Stomp\Symfony\RabbitMqStompTransportFactory;
 use Enqueue\Stomp\Symfony\StompTransportFactory;
@@ -64,5 +66,9 @@ public function build(ContainerBuilder $container)
         if (class_exists(DbalContext::class)) {
             $extension->addTransportFactory(new DbalTransportFactory());
         }
+
+        if (class_exists(SqsContext::class)) {
+            $extension->addTransportFactory(new SqsTransportFactory());
+        }
     }
 }

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

@@ -86,6 +86,16 @@ public function provideEnqueueConfigs()
                     ]
                 ]
             ]],
+            ['sqs' => [
+                'transport' => [
+                    'default' => 'sqs',
+                    'sqs' => [
+                        'key' => getenv('AWS__SQS__KEY'),
+                        'secret' => getenv('AWS__SQS__SECRET'),
+                        'region' => getenv('AWS__SQS__REGION'),
+                    ]
+                ]
+            ]],
         ];
     }
 

pkg/enqueue-bundle/Tests/Unit/EnqueueBundleTest.php

@@ -14,6 +14,7 @@
 use Enqueue\Dbal\Symfony\DbalTransportFactory;
 use Enqueue\Fs\Symfony\FsTransportFactory;
 use Enqueue\Redis\Symfony\RedisTransportFactory;
+use Enqueue\Sqs\Symfony\SqsTransportFactory;
 use Enqueue\Stomp\Symfony\RabbitMqStompTransportFactory;
 use Enqueue\Stomp\Symfony\StompTransportFactory;
 use Enqueue\Symfony\DefaultTransportFactory;
@@ -194,6 +195,23 @@ public function testShouldRegisterDbalTransportFactory()
         $bundle->build($container);
     }
 
+    public function testShouldRegisterSqsTransportFactory()
+    {
+        $extensionMock = $this->createEnqueueExtensionMock();
+
+        $container = new ContainerBuilder();
+        $container->registerExtension($extensionMock);
+
+        $extensionMock
+            ->expects($this->at(9))
+            ->method('addTransportFactory')
+            ->with($this->isInstanceOf(SqsTransportFactory::class))
+        ;
+
+        $bundle = new EnqueueBundle();
+        $bundle->build($container);
+    }
+
     /**
      * @return \PHPUnit_Framework_MockObject_MockObject|EnqueueExtension
      */

pkg/sqs/.gitignore

@@ -0,0 +1,6 @@
+*~
+/composer.lock
+/composer.phar
+/phpunit.xml
+/vendor/
+/.idea/

pkg/sqs/.travis.yml

@@ -0,0 +1,21 @@
+sudo: false
+
+git:
+    depth: 1
+
+language: php
+
+php:
+    - '5.6'
+    - '7.0'
+
+cache:
+    directories:
+        - $HOME/.composer/cache
+
+install:
+    - composer self-update
+    - composer install --prefer-source
+
+script:
+    - vendor/bin/phpunit --exclude-group=functional