[sqs] introduce sqs client that hides differencies of single and multi aws clients.

Author: makasim

docs/transport/sqs.md

@@ -19,6 +19,7 @@ It uses internally official [aws sdk library](https://packagist.org/packages/aws
 * [Send delay message](#send-delay-message)
 * [Consume message](#consume-message)
 * [Purge queue messages](#purge-queue-messages)
+* [Queue from another AWS account](#queue-from-another-aws-account)
 
 ## Installation
 
@@ -122,4 +123,23 @@ $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 `SqsDestination::setQueueOwnerAWSAccountId` method.
+
+```php
+<?php
+use Enqueue\Sqs\SqsConnectionFactory;
+
+// globally for all queues
+$factory = new SqsConnectionFactory('sqs:?queue_owner_aws_account_id=awsAccountId');
+
+$context = (new SqsConnectionFactory('sqs:'))->createContext();
+
+// per queue.
+$queue = $context->createQueue('foo');
+$queue->setQueueOwnerAWSAccountId('awsAccountId');
+```
+
 [back to index](../index.md)

pkg/sqs/SqsClient.php

@@ -0,0 +1,130 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Sqs;
+
+use Aws\MultiRegionClient;
+use Aws\Result;
+use Aws\Sqs\SqsClient as AwsSqsClient;
+
+class SqsClient
+{
+    /**
+     * @var AwsSqsClient
+     */
+    private $singleClient;
+
+    /**
+     * @var MultiRegionClient
+     */
+    private $multiClient;
+
+    /**
+     * @var callable
+     */
+    private $inputClient;
+
+    public function __construct($inputClient)
+    {
+        $this->inputClient = is_callable($inputClient) ? $inputClient : function () use ($inputClient) {
+            return $inputClient;
+        };
+    }
+
+    public function deleteMessage(array $args): Result
+    {
+        return $this->callApi('deleteMessage', $args);
+    }
+
+    public function receiveMessage(array $args): Result
+    {
+        return $this->callApi('receiveMessage', $args);
+    }
+
+    public function purgeQueue(array $args): Result
+    {
+        return $this->callApi('purgeQueue', $args);
+    }
+
+    public function getQueueUrl(array $args): Result
+    {
+        return $this->callApi('getQueueUrl', $args);
+    }
+
+    public function createQueue(array $args): Result
+    {
+        return $this->callApi('createQueue', $args);
+    }
+
+    public function deleteQueue(array $args): Result
+    {
+        return $this->callApi('deleteQueue', $args);
+    }
+
+    public function sendMessage(array $args): Result
+    {
+        return $this->callApi('sendMessage', $args);
+    }
+
+    public function getAWSClient(): AwsSqsClient
+    {
+        $this->resolveClient();
+
+        if ($this->singleClient) {
+            return $this->singleClient;
+        }
+
+        if ($this->multiClient) {
+            $mr = new \ReflectionMethod($this->multiClient, 'getClientFromPool');
+            $mr->setAccessible(true);
+            $singleClient = $mr->invoke($this->multiClient, $this->multiClient->getRegion());
+            $mr->setAccessible(false);
+
+            return $singleClient;
+        }
+
+        throw new \LogicException('The multi or single client must be set');
+    }
+
+    private function callApi(string $name, array $args): Result
+    {
+        if ($this->singleClient) {
+            unset($args['@region']);
+
+            return call_user_func([$this->singleClient, $name], $args);
+        }
+
+        if ($this->multiClient) {
+            return call_user_func([$this->multiClient, $name], $args);
+        }
+
+        throw new \LogicException('The multi or single client must be set');
+    }
+
+    private function resolveClient(): void
+    {
+        if ($this->singleClient || $this->multiClient) {
+            return;
+        }
+
+        $client = call_user_func($this->inputClient);
+        if ($client instanceof MultiRegionClient) {
+            $this->multiClient = $client;
+
+            return;
+        }
+        if ($client instanceof AwsSqsClient) {
+            $this->singleClient = $client;
+
+            return;
+        }
+
+        throw new \LogicException(sprintf(
+            'The input client must be an instance of "%s" or "%s" or a callable that returns one of those. Got "%s"',
+            AwsSqsClient::class,
+            MultiRegionClient::class,
+            is_object($client) ? get_class($client) : gettype($client)
+        ));
+    }
+}

pkg/sqs/SqsConnectionFactory.php

@@ -4,6 +4,8 @@
 
 namespace Enqueue\Sqs;
 
+use Aws\MultiRegionClient;
+use Aws\Sdk;
 use Aws\Sqs\SqsClient;
 use Enqueue\Dsn\Dsn;
 use Interop\Queue\ConnectionFactory;
@@ -17,7 +19,7 @@ class SqsConnectionFactory implements ConnectionFactory
     private $config;
 
     /**
-     * @var SqsClient
+     * @var MultiRegionClient
      */
     private $client;
 
@@ -81,7 +83,7 @@ public function createContext(): Context
         return new SqsContext($this->establishConnection(), $this->config);
     }
 
-    private function establishConnection(): SqsClient
+    private function establishConnection(): MultiRegionClient
     {
         if ($this->client) {
             return $this->client;
@@ -108,7 +110,7 @@ private function establishConnection(): SqsClient
             }
         }
 
-        $this->client = new SqsClient($config);
+        $this->client = (new Sdk(['Sqs' => $config]))->createMultiRegionSqs();
 
         return $this->client;
     }

pkg/sqs/SqsConsumer.php

@@ -120,6 +120,7 @@ public function acknowledge(Message $message): void
         InvalidMessageException::assertMessageInstanceOf($message, SqsMessage::class);
 
         $this->context->getClient()->deleteMessage([
+            '@region' => $this->queue->getRegion(),
             'QueueUrl' => $this->context->getQueueUrl($this->queue),
             'ReceiptHandle' => $message->getReceiptHandle(),
         ]);
@@ -133,6 +134,7 @@ public function reject(Message $message, bool $requeue = false): void
         InvalidMessageException::assertMessageInstanceOf($message, SqsMessage::class);
 
         $this->context->getClient()->deleteMessage([
+            '@region' => $this->queue->getRegion(),
             'QueueUrl' => $this->context->getQueueUrl($this->queue),
             'ReceiptHandle' => $message->getReceiptHandle(),
         ]);
@@ -149,6 +151,7 @@ protected function receiveMessage(int $timeoutSeconds): ?SqsMessage
         }
 
         $arguments = [
+            '@region' => $this->queue->getRegion(),
             'AttributeNames' => ['All'],
             'MessageAttributeNames' => ['All'],
             'MaxNumberOfMessages' => $this->maxNumberOfMessages,

pkg/sqs/SqsContext.php

@@ -119,6 +119,7 @@ public function purgeQueue(Queue $queue): void
         InvalidDestinationException::assertDestinationInstanceOf($queue, SqsDestination::class);
 
         $this->getClient()->purgeQueue([
+            '@region' => $queue->getRegion(),
             'QueueUrl' => $this->getQueueUrl($queue),
         ]);
     }
@@ -152,8 +153,14 @@ public function getQueueUrl(SqsDestination $destination): string
             return $this->queueUrls[$destination->getQueueName()];
         }
 
-        $arguments = ['QueueName' => $destination->getQueueName()];
-        if (false == empty($this->config['queue_owner_aws_account_id'])) {
+        $arguments = [
+            '@region' => $destination->getRegion(),
+            'QueueName' => $destination->getQueueName()
+        ];
+
+        if ($destination->getQueueOwnerAWSAccountId()) {
+            $arguments['QueueOwnerAWSAccountId'] = $destination->getQueueOwnerAWSAccountId();
+        } elseif (false == empty($this->config['queue_owner_aws_account_id'])) {
             $arguments['QueueOwnerAWSAccountId'] = $this->config['queue_owner_aws_account_id'];
         }
 
@@ -169,6 +176,7 @@ public function getQueueUrl(SqsDestination $destination): string
     public function declareQueue(SqsDestination $dest): void
     {
         $result = $this->getClient()->createQueue([
+            '@region' => $dest->getRegion(),
             'Attributes' => $dest->getAttributes(),
             'QueueName' => $dest->getQueueName(),
         ]);

pkg/sqs/SqsDestination.php

@@ -14,11 +14,21 @@ class SqsDestination implements Topic, Queue
      */
     private $name;
 
+    /**
+     * @var string|null
+     */
+    private $region;
+
     /**
      * @var array
      */
     private $attributes;
 
+    /**
+     * @var string|null
+     */
+    private $queueOwnerAWSAccountId;
+
     /**
      * The name of the new queue.
      * The following limits apply to this name:
@@ -187,4 +197,24 @@ public function setContentBasedDeduplication(bool $enable): void
             unset($this->attributes['ContentBasedDeduplication']);
         }
     }
+
+    public function getQueueOwnerAWSAccountId(): ?string
+    {
+        return $this->queueOwnerAWSAccountId;
+    }
+
+    public function setQueueOwnerAWSAccountId(?string $queueOwnerAWSAccountId): void
+    {
+        $this->queueOwnerAWSAccountId = $queueOwnerAWSAccountId;
+    }
+
+    public function setRegion(string $region = null): void
+    {
+        $this->region = $region;
+    }
+
+    public function getRegion(): ?string
+    {
+        return $this->region;
+    }
 }

pkg/sqs/SqsProducer.php

@@ -44,6 +44,7 @@ public function send(Destination $destination, Message $message): void
         }
 
         $arguments = [
+            '@region' => $destination->getRegion(),
             'MessageAttributes' => [
                 'Headers' => [
                     'DataType' => 'String',

pkg/sqs/Tests/Spec/SqsSendToAndReceiveFromQueueTest.php

@@ -4,15 +4,18 @@
 
 use Enqueue\Sqs\SqsContext;
 use Enqueue\Sqs\SqsDestination;
+use Enqueue\Test\RetryTrait;
 use Enqueue\Test\SqsExtension;
 use Interop\Queue\Context;
 use Interop\Queue\Spec\SendToAndReceiveFromQueueSpec;
 
 /**
  * @group functional
+ * @retry 5
  */
 class SqsSendToAndReceiveFromQueueTest extends SendToAndReceiveFromQueueSpec
 {
+    use RetryTrait;
     use SqsExtension;
     use CreateSqsQueueTrait;
 

pkg/sqs/Tests/Spec/SqsSendToAndReceiveNoWaitFromQueueTest.php

@@ -4,15 +4,18 @@
 
 use Enqueue\Sqs\SqsContext;
 use Enqueue\Sqs\SqsDestination;
+use Enqueue\Test\RetryTrait;
 use Enqueue\Test\SqsExtension;
 use Interop\Queue\Context;
 use Interop\Queue\Spec\SendToAndReceiveNoWaitFromQueueSpec;
 
 /**
  * @group functional
+ * @retry 5
  */
 class SqsSendToAndReceiveNoWaitFromQueueTest extends SendToAndReceiveNoWaitFromQueueSpec
 {
+    use RetryTrait;
     use SqsExtension;
     use CreateSqsQueueTrait;
 

pkg/sqs/Tests/Spec/SqsSendToAndReceiveNoWaitFromTopicTest.php

@@ -4,15 +4,18 @@
 
 use Enqueue\Sqs\SqsContext;
 use Enqueue\Sqs\SqsDestination;
+use Enqueue\Test\RetryTrait;
 use Enqueue\Test\SqsExtension;
 use Interop\Queue\Context;
 use Interop\Queue\Spec\SendToAndReceiveNoWaitFromTopicSpec;
 
 /**
  * @group functional
+ * @retry 5
  */
 class SqsSendToAndReceiveNoWaitFromTopicTest extends SendToAndReceiveNoWaitFromTopicSpec
 {
+    use RetryTrait;
     use SqsExtension;
     use CreateSqsQueueTrait;