Fix multiple subscriber issue

Author: munir131

.gitignore

@@ -1,2 +1,3 @@
-vendor
-composer.lock
+vendor
+composer.lock
+.DS_store

src/Jobs/PubSubJob.php

@@ -2,11 +2,11 @@
 
 namespace PubSub\PubSubQueue\Jobs;
 
-use Illuminate\Queue\Jobs\Job;
 use Google\Cloud\PubSub\Message;
 use Illuminate\Container\Container;
-use PubSub\PubSubQueue\PubSubQueue;
 use Illuminate\Contracts\Queue\Job as JobContract;
+use Illuminate\Queue\Jobs\Job;
+use PubSub\PubSubQueue\PubSubQueue;
 
 class PubSubJob extends Job implements JobContract
 {
@@ -24,6 +24,13 @@ class PubSubJob extends Job implements JobContract
      */
     protected $job;
 
+    /**
+     * subscriber name
+     *
+     * @var string
+     */
+    protected $subscriber;
+
     /**
      * Create a new job instance.
      *
@@ -33,14 +40,14 @@ class PubSubJob extends Job implements JobContract
      * @param string       $connectionName
      * @param string       $queue
      */
-    public function __construct(Container $container, PubSubQueue $pubsub, Message $job, $connectionName, $queue)
+    public function __construct(Container $container, PubSubQueue $pubsub, Message $job, $connectionName, $queue, $subscriber = null)
     {
         $this->pubsub = $pubsub;
         $this->job = $job;
         $this->queue = $queue;
         $this->container = $container;
         $this->connectionName = $connectionName;
-
+        $this->subscriber = $subscriber;
         $this->decoded = $this->payload();
     }
 

src/PubSubQueue.php

@@ -2,13 +2,13 @@
 
 namespace PubSub\PubSubQueue;
 
-use Illuminate\Queue\Queue;
-use Illuminate\Support\Str;
-use Google\Cloud\PubSub\Topic;
 use Google\Cloud\PubSub\Message;
 use Google\Cloud\PubSub\PubSubClient;
-use PubSub\PubSubQueue\Jobs\PubSubJob;
+use Google\Cloud\PubSub\Topic;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
+use Illuminate\Queue\Queue;
+use Illuminate\Support\Str;
+use PubSub\PubSubQueue\Jobs\PubSubJob;
 
 class PubSubQueue extends Queue implements QueueContract
 {
@@ -31,6 +31,10 @@ class PubSubQueue extends Queue implements QueueContract
      */
     protected $config;
 
+    /**
+     * Subscriber name
+     */
+    protected $subscriber;
 
     /**
      * Create a new GCP PubSub instance.
@@ -86,11 +90,9 @@ public function pushRaw($payload, $subscriber = null, array $options = [])
     {
         $topic = $this->getTopic($subscriber, true);
 
-        $this->subscribeToTopic($topic);
-
         $publish = ['data' => $payload];
 
-        if (! empty($options)) {
+        if (!empty($options)) {
             $publish['attributes'] = $options;
         }
 
@@ -128,25 +130,26 @@ public function later($delay, $job, $data = '', $subscriber = null)
      */
     public function pop($subscriber = null)
     {
+        $this->subscriber = $subscriber;
         $topic = $this->getTopic($this->getQueue($subscriber));
 
-        if (! $topic->exists()) {
+        if (!$topic->exists()) {
             return;
         }
-
-        $subscription = $topic->subscription($this->getSubscriberName($subscriber));
+        $subscription = $topic->subscription($subscriber);
         $messages = $subscription->pull([
             'returnImmediately' => true,
             'maxMessages' => 1,
         ]);
 
-        if (! empty($messages) && count($messages) > 0) {
+        if (!empty($messages) && count($messages) > 0) {
             return new PubSubJob(
                 $this->container,
                 $this,
                 $messages[0],
                 $this->connectionName,
-                $this->getQueue($subscriber)
+                $this->getQueue($subscriber),
+                $subscriber
             );
         }
     }
@@ -170,8 +173,6 @@ public function bulk($jobs, $data = '', $queue = null)
 
         $topic = $this->getTopic($this->getQueue($queue), true);
 
-        $this->subscribeToTopic($topic);
-
         return $topic->publishBatch($payloads);
     }
 
@@ -183,7 +184,7 @@ public function bulk($jobs, $data = '', $queue = null)
      */
     public function acknowledge(Message $message, $queue = null)
     {
-        $subscription = $this->getTopic($this->getQueue($queue))->subscription($this->getSubscriberName($queue));
+        $subscription = $this->getTopic($this->getQueue($queue))->subscription($this->subscriber);
         $subscription->acknowledge($message);
     }
 
@@ -198,7 +199,7 @@ public function acknowledge(Message $message, $queue = null)
     public function acknowledgeAndPublish(Message $message, $queue = null, $options = [], $delay = 0)
     {
         $topic = $this->getTopic($this->getQueue($queue));
-        $subscription = $topic->subscription($this->getSubscriberName($queue));
+        $subscription = $topic->subscription($queue);
 
         $subscription->acknowledge($message);
 
@@ -257,7 +258,7 @@ public function getTopic($queue, $create = false)
         $queue = $this->getQueue($queue);
         $topic = $this->pubsub->topic($queue);
 
-        if (! $topic->exists() && $create) {
+        if (!$topic->exists() && $create) {
             $topic->create();
         }
 
@@ -271,11 +272,11 @@ public function getTopic($queue, $create = false)
      *
      * @return \Google\Cloud\PubSub\Subscription
      */
-    public function subscribeToTopic(Topic $topic)
+    public function subscribeToTopic(Topic $topic, $subscriber = null)
     {
-        $subscription = $topic->subscription($this->getSubscriberName());
-        if (! $subscription->exists()) {
-            $subscription = $topic->subscribe($this->getSubscriberName());
+        $subscription = $topic->subscription($subscriber);
+        if (!$subscription->exists()) {
+            $subscription = $topic->subscribe($subscriber);
         }
 
         return $subscription;
@@ -290,10 +291,7 @@ public function subscribeToTopic(Topic $topic)
      */
     public function getSubscriberName($queue = null)
     {
-        if ($this->config && $this->config['subscribers'] && $queue && isset($this->config['subscribers'][$queue])) {
-            return $this->config['subscribers'][$queue];
-        }
-        return 'subscriber';
+        return $queue;
     }
 
     /**
@@ -314,7 +312,10 @@ public function getPubSub()
      */
     public function getQueue($queue)
     {
-        return $queue ?: $this->default;
+        if ($this->config && $this->config['subscribers'] && $queue && isset($this->config['subscribers'][$queue])) {
+            return $this->config['subscribers'][$queue];
+        }
+        return $this->default;
     }
 
     /**

tests/Unit/PubSubQueueTests.php

@@ -312,11 +312,6 @@ public function testSubscriptionIsRetrieved()
         $this->assertTrue($queue->subscribeToTopic($this->topic) instanceof Subscription);
     }
 
-    public function testGetSubscriberName()
-    {
-        $this->assertTrue(is_string($this->queue->getSubscriberName()));
-    }
-
     public function testGetPubSub()
     {
         $this->assertTrue($this->queue->getPubSub() instanceof PubSubClient);