snsqs client driver

ASKozienko · ASKozienko · commit 8c0e3c62d2a0 · 2019-02-19T15:35:48.000+02:00
diff --git a/phpunit.xml.dist b/phpunit.xml.dist
@@ -65,6 +65,10 @@
             <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>
diff --git a/pkg/enqueue/Client/Driver/SnsQsDriver.php b/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);
+    }
+}
diff --git a/pkg/enqueue/Client/Resources.php b/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,
diff --git a/pkg/enqueue/Resources.php b/pkg/enqueue/Resources.php
@@ -155,6 +155,10 @@ public static function getKnownConnections(): array
                 'schemes' => ['sqs'],
                 'supportedSchemeExtensions' => [],
                 'package' => 'enqueue/sqs', ];
+            $map[SqsConnectionFactory::class] = [
+                'schemes' => ['snsqs'],
+                'supportedSchemeExtensions' => [],
+                'package' => 'enqueue/snsqs', ];
             $map[GpsConnectionFactory::class] = [
                 'schemes' => ['gps'],
                 'supportedSchemeExtensions' => [],
