Add prefixes to queue names. (#46)

* Optional subscription and topic creation (#35)

* ability to disable automatic topic creation and validation
which cause high usage of "Administrative operations" calls and can lead to Quota limit exceptions

* ability to disable automatic subscription
which cause high usage of "Administrative operations" calls and can lead to Quota limit exceptions

* style: styleci fix

* Add ability to specify queue prefixes.

* Fix formatting.

* Update README to include queue_prefix config.

Co-authored-by: Maksim Martianov <[email protected]>

Author: garbetjie

README.md

@@ -29,6 +29,7 @@ You can check [Google Cloud PubSub client](http://googleapis.github.io/google-cl
 'pubsub' => [
     'driver' => 'pubsub',
     'queue' => env('PUBSUB_QUEUE', 'default'),
+    'queue_prefix' => env('PUBSUB_QUEUE_PREFIX', ''),
     'project_id' => env('PUBSUB_PROJECT_ID', 'your-project-id'),
     'retries' => 3,
     'request_timeout' => 60,

src/Connectors/PubSubConnector.php

@@ -29,7 +29,10 @@ public function connect(array $config)
         return new PubSubQueue(
             new PubSubClient($gcp_config),
             $config['queue'] ?? $this->default_queue,
-            $config['subscriber'] ?? 'subscriber'
+            $config['subscriber'] ?? 'subscriber',
+            $config['create_topics'] ?? true,
+            $config['create_subscriptions'] ?? true,
+            $config['queue_prefix'] ?? ''
         );
     }
 

src/PubSubQueue.php

@@ -33,17 +33,41 @@ class PubSubQueue extends Queue implements QueueContract
      */
     protected $subscriber;
 
+    /**
+     * Create topics automatically.
+     *
+     * @var bool
+     */
+    protected $topicAutoCreation;
+
+    /**
+     * Create subscriptions automatically.
+     *
+     * @var bool
+     */
+    protected $subscriptionAutoCreation;
+
+    /**
+     * Prepend all queue names with this prefix.
+     *
+     * @var string
+     */
+    protected $queuePrefix = '';
+
     /**
      * Create a new GCP PubSub instance.
      *
      * @param \Google\Cloud\PubSub\PubSubClient $pubsub
      * @param string $default
      */
-    public function __construct(PubSubClient $pubsub, $default, $subscriber = 'subscriber')
+    public function __construct(PubSubClient $pubsub, $default, $subscriber = 'subscriber', $topicAutoCreation = true, $subscriptionAutoCreation = true, $queuePrefix = '')
     {
         $this->pubsub = $pubsub;
         $this->default = $default;
         $this->subscriber = $subscriber;
+        $this->topicAutoCreation = $topicAutoCreation;
+        $this->subscriptionAutoCreation = $subscriptionAutoCreation;
+        $this->queuePrefix = $queuePrefix;
     }
 
     /**
@@ -85,7 +109,7 @@ public function push($job, $data = '', $queue = null)
      */
     public function pushRaw($payload, $queue = null, array $options = [])
     {
-        $topic = $this->getTopic($queue, true);
+        $topic = $this->getTopic($queue, $this->topicAutoCreation);
 
         $this->subscribeToTopic($topic);
 
@@ -131,7 +155,7 @@ public function pop($queue = null)
     {
         $topic = $this->getTopic($this->getQueue($queue));
 
-        if (! $topic->exists()) {
+        if ($this->topicAutoCreation && ! $topic->exists()) {
             return;
         }
 
@@ -179,7 +203,7 @@ public function bulk($jobs, $data = '', $queue = null)
             $payloads[] = ['data' => base64_encode($payload)];
         }
 
-        $topic = $this->getTopic($this->getQueue($queue), true);
+        $topic = $this->getTopic($this->getQueue($queue), $this->topicAutoCreation);
 
         $this->subscribeToTopic($topic);
 
@@ -274,7 +298,8 @@ public function getTopic($queue, $create = false)
         $queue = $this->getQueue($queue);
         $topic = $this->pubsub->topic($queue);
 
-        if (! $topic->exists() && $create) {
+        // don't check topic if automatic creation is not required, to avoid additional administrator operations calls
+        if ($create && ! $topic->exists()) {
             $topic->create();
         }
 
@@ -292,7 +317,8 @@ public function subscribeToTopic(Topic $topic)
     {
         $subscription = $topic->subscription($this->getSubscriberName());
 
-        if (! $subscription->exists()) {
+        // don't check subscription if automatic creation is not required, to avoid additional administrator operations calls
+        if ($this->subscriptionAutoCreation && ! $subscription->exists()) {
             $subscription = $topic->subscribe($this->getSubscriberName());
         }
 
@@ -329,7 +355,13 @@ public function getPubSub()
      */
     public function getQueue($queue)
     {
-        return $queue ?: $this->default;
+        $queue = $queue ?: $this->default;
+
+        if (! $this->queuePrefix || Str::startsWith($queue, $this->queuePrefix)) {
+            return $queue;
+        }
+
+        return $this->queuePrefix.$queue;
     }
 
     /**

tests/Unit/Connectors/PubSubConnectorTests.php

@@ -27,6 +27,24 @@ public function testConnectReturnsPubSubQueueInstance()
         $this->assertEquals($queue->getSubscriberName(), 'test-subscriber');
     }
 
+    public function testQueuePrefixAdded()
+    {
+        $connector = new PubSubConnector();
+        $config = $this->createFakeConfig() + ['queue_prefix' => 'prefix-'];
+        $queue = $connector->connect($config);
+
+        $this->assertEquals('prefix-my-queue', $queue->getQueue('my-queue'));
+    }
+
+    public function testNotQueuePrefixAddedMultipleTimes()
+    {
+        $connector = new PubSubConnector();
+        $config = $this->createFakeConfig() + ['queue_prefix' => 'prefix-'];
+        $queue = $connector->connect($config);
+
+        $this->assertEquals('prefix-default', $queue->getQueue($queue->getQueue('default')));
+    }
+
     private function createFakeConfig()
     {
         return [