Initial commit

Author: marickvantuil

.gitignore

@@ -0,0 +1,3 @@
+/vendor/
+composer.lock
+.idea/

composer.json

@@ -0,0 +1,31 @@
+{
+    "name": "stackkit/laravel-google-pubsub-queue",
+    "license": "MIT",
+    "authors": [
+        {
+            "name": "Marick van Tuil",
+            "email": "info@marickvantuil.nl"
+        }
+    ],
+    "require": {
+        "ext-json": "*",
+        "google/cloud-pubsub": "^1.24"
+    },
+    "require-dev": {
+        "mockery/mockery": "^1.2",
+        "orchestra/testbench": "^3.5 || ^3.6 || ^3.7 || ^3.8 || ^4.0 || ^5.0",
+        "symfony/console": "^4.4|^5.0"
+    },
+    "autoload": {
+        "psr-4": {
+            "Stackkit\\LaravelGooglePubSubQueue\\": "src/"
+        }
+    },
+    "extra": {
+        "laravel": {
+            "providers": [
+                "Stackkit\\LaravelGooglePubSubQueue\\LaravelGooglePubSubQueueServiceProvider"
+            ]
+        }
+    }
+}

src/LaravelGooglePubSubQueueServiceProvider.php

@@ -0,0 +1,51 @@
+<?php
+
+namespace Stackkit\LaravelGooglePubSubQueue;
+
+use Illuminate\Support\Facades\Route;
+use Illuminate\Support\ServiceProvider;
+use Google\Cloud\PubSub\PubSubClient;
+
+class LaravelGooglePubSubQueueServiceProvider extends ServiceProvider
+{
+    /**
+     * Perform post-registration booting of services.
+     *
+     * @return void
+     */
+    public function boot()
+    {
+        $this->app['queue']->addConnector('pubsub', function () {
+            return new PubSubConnector();
+        });
+
+        Route::post('pubsub-wake', function () {
+            $client = new PubSubClient([
+                "driver" => "pubsub",
+                "connection" => "default",
+                "queue" => "projects/test-marick/topics/geencijfer-prd-queue",
+                "project_id" => "test-marick",
+                "retries" => 3,
+                "request_timeout" => 60,
+                "subscriber" => "projects/test-marick/subscriptions/geencijfer-prd-queue-subscription",
+                "keyFilePath" => "/var/www/gcloud-key.json",
+            ]);
+
+            $message = $client->consume(request()->toArray());
+
+            logger(base64_decode($message->data()));
+        });
+    }
+
+    /**
+     * Register the service provider.
+     *
+     * @return void
+     */
+    public function register()
+    {
+//        $this->commands([
+//            SendEmailsCommand::class,
+//        ]);
+    }
+}

src/PubSubConnector.php

@@ -0,0 +1,18 @@
+<?php
+
+namespace Stackkit\LaravelGooglePubSubQueue;
+
+use Google\Cloud\PubSub\PubSubClient;
+use Illuminate\Queue\Connectors\ConnectorInterface;
+
+class PubSubConnector implements ConnectorInterface
+{
+    public function connect(array $config)
+    {
+        return new PubSubQueue(
+            new PubSubClient($config),
+            $config['queue'] ,
+            $config['subscriber']
+        );
+    }
+}

src/PubSubJob.php

@@ -0,0 +1,95 @@
+<?php
+
+namespace Stackkit\LaravelGooglePubSubQueue;
+
+use Google\Cloud\PubSub\Message;
+use Illuminate\Container\Container;
+use Illuminate\Contracts\Queue\Job as JobContract;
+use Illuminate\Queue\Jobs\Job;
+use Stackkit\LaravelGooglePubSubQueue\PubSubQueue;
+
+class PubSubJob extends Job implements JobContract
+{
+    /**
+     * The PubSub queue.
+     *
+     * @var \Stackkit\LaravelGooglePubSubQueue\PubSubQueue
+     */
+    protected $pubsub;
+
+    /**
+     * The job instance.
+     *
+     * @var array
+     */
+    protected $job;
+
+    /**
+     * Create a new job instance.
+     *
+     * @param \Illuminate\Container\Container $container
+     * @param \Stackkit\LaravelGooglePubSubQueue\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)
+    {
+        $this->pubsub = $pubsub;
+        $this->job = $job;
+        $this->queue = $queue;
+        $this->container = $container;
+        $this->connectionName = $connectionName;
+
+        $this->decoded = $this->payload();
+    }
+
+    /**
+     * Get the job identifier.
+     *
+     * @return string
+     */
+    public function getJobId()
+    {
+        return $this->decoded['id'] ?? null;
+    }
+
+    /**
+     * Get the raw body of the job.
+     *
+     * @return string
+     */
+    public function getRawBody()
+    {
+        return base64_decode($this->job->data());
+    }
+
+    /**
+     * Get the number of times the job has been attempted.
+     *
+     * @return int
+     */
+    public function attempts()
+    {
+        return ((int) $this->job->attribute('attempts') ?? 0) + 1;
+    }
+
+    /**
+     * Release the job back into the queue.
+     *
+     * @param  int   $delay
+     * @return void
+     */
+    public function release($delay = 0)
+    {
+        parent::release($delay);
+
+        $attempts = $this->attempts();
+        $this->pubsub->republish(
+            $this->job,
+            $this->queue,
+            ['attempts' => (string) $attempts],
+            $delay
+        );
+    }
+}