Merge branch '6.0'

Author: munir131

.gitignore

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

composer.json

@@ -1,8 +1,8 @@
 {
-    "name": "kainxspirits/laravel-pubsub-queue",
+    "name": "munir131/laravel-pubsub-queue",
     "description": "Queue driver for Google Cloud Pub/Sub.",
     "keywords": [
-        "kainxspirits",
+        "munir131",
         "laravel",
         "queue",
         "gcp",
@@ -13,26 +13,27 @@
     "type": "library",
     "authors": [
         {
-            "name": "Kendryck",
-            "email": "kainxspirits@users.noreply.github.com"
+            "name": "Munir Khakhi",
+            "email": "munir131@users.noreply.github.com"
         }
     ],
     "require": {
-        "php" : ">=7.1",
-        "illuminate/queue": "~5.7.0",
-        "google/cloud-pubsub": "^1.1"
+        "php": ">=7.1",
+        "google/cloud-pubsub": "^1.1",
+        "illuminate/queue": "6.*",
+        "laravel/helpers": "^1.1"
     },
     "require-dev": {
         "phpunit/phpunit": "^7.1"
     },
     "autoload": {
         "psr-4": {
-            "Kainxspirits\\PubSubQueue\\": "src/"
+            "PubSub\\PubSubQueue\\": "src/"
         }
     },
     "autoload-dev": {
         "psr-4": {
-            "Kainxspirits\\PubSubQueue\\Tests\\": "tests/"
+            "PubSub\\PubSubQueue\\Tests\\": "tests/"
         }
     },
     "scripts": {
@@ -41,7 +42,7 @@
     "extra": {
         "laravel": {
             "providers": [
-                "Kainxspirits\\PubSubQueue\\PubSubQueueServiceProvider"
+                "PubSub\\PubSubQueue\\PubSubQueueServiceProvider"
             ]
         }
     },

src/Connectors/PubSubConnector.php

@@ -1,9 +1,9 @@
 <?php
 
-namespace Kainxspirits\PubSubQueue\Connectors;
+namespace PubSub\PubSubQueue\Connectors;
 
 use Google\Cloud\PubSub\PubSubClient;
-use Kainxspirits\PubSubQueue\PubSubQueue;
+use PubSub\PubSubQueue\PubSubQueue;
 use Illuminate\Queue\Connectors\ConnectorInterface;
 
 class PubSubConnector implements ConnectorInterface
@@ -24,10 +24,10 @@ class PubSubConnector implements ConnectorInterface
     public function connect(array $config)
     {
         $gcp_config = $this->transformConfig($config);
-
         return new PubSubQueue(
             new PubSubClient($gcp_config),
-            $config['queue'] ?? $this->default_queue
+            $config['queue'] ?? $this->default_queue,
+            $gcp_config
         );
     }
 

src/Jobs/PubSubJob.php

@@ -1,19 +1,19 @@
 <?php
 
-namespace Kainxspirits\PubSubQueue\Jobs;
+namespace PubSub\PubSubQueue\Jobs;
 
-use Illuminate\Queue\Jobs\Job;
 use Google\Cloud\PubSub\Message;
 use Illuminate\Container\Container;
-use Kainxspirits\PubSubQueue\PubSubQueue;
 use Illuminate\Contracts\Queue\Job as JobContract;
+use Illuminate\Queue\Jobs\Job;
+use PubSub\PubSubQueue\PubSubQueue;
 
 class PubSubJob extends Job implements JobContract
 {
     /**
      * The PubSub queue.
      *
-     * @var \Kainxspirits\PubSubQueue\PubSubQueue
+     * @var \PubSub\PubSubQueue\PubSubQueue
      */
     protected $pubsub;
 
@@ -24,23 +24,30 @@ class PubSubJob extends Job implements JobContract
      */
     protected $job;
 
+    /**
+     * subscriber name
+     *
+     * @var string
+     */
+    protected $subscriber;
+
     /**
      * Create a new job instance.
      *
      * @param \Illuminate\Container\Container $container
-     * @param \Kainxspirits\PubSubQueue\PubSubQueue $sqs
+     * @param \PubSub\PubSubQueue\PubSubQueue $sqs
      * @param \Google\Cloud\PubSub\Message $job
      * @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, $subscriberName, $subscriber = null)
     {
         $this->pubsub = $pubsub;
         $this->job = $job;
-        $this->queue = $queue;
+        $this->queue = $subscriberName;
         $this->container = $container;
         $this->connectionName = $connectionName;
-
+        $this->subscriber = $subscriber;
         $this->decoded = $this->payload();
     }
 
@@ -61,9 +68,34 @@ public function getJobId()
      */
     public function getRawBody()
     {
+        if ($this->pubsub->checkHandler($this->subscriber)) {
+            return $this->modifyPayload(
+                $this->job->data(),
+                $this->pubsub->getHandler($this->subscriber)
+            );
+        }
         return base64_decode($this->job->data());
     }
 
+    /**
+     * @param string|array $payload
+     * @param string $class
+     * @return array
+     */
+    private function modifyPayload($payload, $class)
+    {
+        if (!is_array($payload)) {
+            $payload = json_decode($payload, true);
+        }
+
+        $body = [
+            'job' => $class . '@handle',
+            'data' => $payload,
+        ];
+
+        return json_encode($body);
+    }
+
     /**
      * Delete the job from the queue.
      *