WIP

Author: marickvantuil

src/CloudTasksConnector.php

@@ -2,12 +2,23 @@
 
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
+use Google\Cloud\Tasks\V2\CloudTasksClient;
 use Illuminate\Queue\Connectors\ConnectorInterface;
 
 class CloudTasksConnector implements ConnectorInterface
 {
     public function connect(array $config)
     {
-        return new CloudTasksQueue();
+        return new CloudTasksQueue(
+            $config,
+            new CloudTasksClient($this->buildConfig())
+        );
+    }
+
+    private function buildConfig()
+    {
+        return [
+            'credentials' => Config::credentials(),
+        ];
     }
 }

src/CloudTasksException.php

@@ -0,0 +1,10 @@
+<?php
+
+namespace Stackkit\LaravelGoogleCloudTasksQueue;
+
+use Exception;
+
+class CloudTasksException extends Exception
+{
+    //
+}

src/CloudTasksJob.php

@@ -2,23 +2,34 @@
 
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
+use Illuminate\Container\Container;
 use Illuminate\Queue\Jobs\Job as LaravelJob;
 use Illuminate\Contracts\Queue\Job as JobContract;
 
 class CloudTasksJob extends LaravelJob implements JobContract
 {
+    private $job;
+    private $attempts;
+
+    public function __construct($job, $attempts)
+    {
+        $this->job = $job;
+        $this->attempts = $attempts;
+        $this->container = Container::getInstance();
+    }
+
     public function getJobId()
     {
-        // TODO: Implement getJobId() method.
+        return $this->job['uuid'];
     }
 
     public function getRawBody()
     {
-        // TODO: Implement getRawBody() method.
+        return json_encode($this->job);
     }
 
     public function attempts()
     {
-        // TODO: Implement attempts() method.
+        return $this->attempts;
     }
 }

src/CloudTasksQueue.php

@@ -2,33 +2,81 @@
 
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
+use Carbon\Carbon;
+use Google\Cloud\Tasks\V2\CloudTasksClient;
+use Google\Cloud\Tasks\V2\HttpMethod;
+use Google\Cloud\Tasks\V2\HttpRequest;
+use Google\Cloud\Tasks\V2\Task;
+use Google\Protobuf\Timestamp;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
 use Illuminate\Queue\Queue as LaravelQueue;
+use Illuminate\Support\InteractsWithTime;
 
 class CloudTasksQueue extends LaravelQueue implements QueueContract
 {
+    use InteractsWithTime;
+
+    private $client;
+    private $default;
+
+    public function __construct(array $config, CloudTasksClient $client)
+    {
+        $this->client = $client;
+        $this->default = $config['queue'];
+    }
+
     public function size($queue = null)
     {
         // TODO: Implement size() method.
     }
 
     public function push($job, $data = '', $queue = null)
     {
-        // TODO: Implement push() method.
+        return $this->pushToCloudTasks($queue, $this->createPayload(
+            $job, $this->getQueue($queue), $data
+        ));
     }
 
     public function pushRaw($payload, $queue = null, array $options = [])
     {
-        // TODO: Implement pushRaw() method.
+        return $this->pushToCloudTasks($queue, $payload);
     }
 
     public function later($delay, $job, $data = '', $queue = null)
     {
-        // TODO: Implement later() method.
+        return $this->pushToCloudTasks($queue, $this->createPayload(
+            $job, $this->getQueue($queue), $data
+        ), $delay);
+    }
+
+    protected function pushToCloudTasks($queue, $payload, $delay = 0, $attempts = 0)
+    {
+        $queue = $this->getQueue($queue);
+        $queueName = $this->client->queueName(Config::project(), Config::location(), $queue);
+        $availableAt = $this->availableAt($delay);
+
+        $httpRequest = new HttpRequest();
+        $httpRequest->setUrl(Config::handler());
+        $httpRequest->setHttpMethod(HttpMethod::POST);
+        $httpRequest->setBody($payload);
+
+        $task = new Task;
+        $task->setHttpRequest($httpRequest);
+
+        if ($availableAt > time()) {
+            $task->setScheduleTime(new Timestamp(['seconds' => $availableAt]));
+        }
+
+        $this->client->createTask($queueName, $task);
     }
 
     public function pop($queue = null)
     {
         // TODO: Implement pop() method.
     }
+
+    private function getQueue($queue = null)
+    {
+        return $queue ?: $this->default;
+    }
 }

src/CloudTasksServiceProvider.php

@@ -2,14 +2,27 @@
 
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
+use Illuminate\Queue\QueueManager;
+use Illuminate\Routing\Router;
 use Illuminate\Support\ServiceProvider as LaravelServiceProvider;
 
 class CloudTasksServiceProvider extends LaravelServiceProvider
 {
-    public function boot()
+    public function boot(QueueManager $queue, Router $router)
     {
-        $this->app['queue']->addConnector('cloudtasks', function () {
+        $this->registerConnector($queue);
+        $this->registerRoutes($router);
+    }
+
+    private function registerConnector(QueueManager $queue)
+    {
+        $queue->addConnector('cloudtasks', function () {
             return new CloudTasksConnector;
         });
     }
+
+    private function registerRoutes(Router $router)
+    {
+        $router->post('handle-task', [TaskHandler::class, 'handle']);
+    }
 }

src/Config.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace Stackkit\LaravelGoogleCloudTasksQueue;
+
+class Config
+{
+    public static function credentials()
+    {
+        return config('queue.connections.cloudtasks.credentials');
+    }
+
+    public static function project()
+    {
+        return config('queue.connections.cloudtasks.project');
+    }
+
+    public static function location()
+    {
+        return config('queue.connections.cloudtasks.location');
+    }
+
+    public static function handler()
+    {
+        return config('queue.connections.cloudtasks.handler');
+    }
+}

src/TaskHandler.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace Stackkit\LaravelGoogleCloudTasksQueue;
+
+use Illuminate\Queue\Worker;
+use Illuminate\Queue\WorkerOptions;
+use Throwable;
+
+class TaskHandler
+{
+    /**
+     * @throws CloudTasksException
+     */
+    public function handle()
+    {
+        $task = $this->captureTask();
+
+        $this->handleTask($task);
+    }
+
+    /**
+     * @throws CloudTasksException
+     */
+    private function captureTask()
+    {
+        $input = file_get_contents('php://input');
+
+        if ($input === false) {
+            throw new CloudTasksException('Could not read incoming task');
+        }
+
+        $task = json_decode($input, true, JSON_THROW_ON_ERROR);
+
+        if (is_null($task)) {
+            throw new CloudTasksException('Could not decode incoming task');
+        }
+
+        return $task;
+    }
+
+    /**
+     * @param $task
+     * @throws CloudTasksException
+     */
+    private function handleTask($task)
+    {
+        $job = new CloudTasksJob($task, request()->header('X-CloudTasks-TaskRetryCount'));
+
+        $worker = $this->getQueueWorker();
+
+        try {
+            $worker->process('cloudtasks', $job, new WorkerOptions());
+        } catch (Throwable $e) {
+            throw new CloudTasksException('Failed handling task');
+        }
+    }
+
+    /**
+     * @return Worker
+     */
+    private function getQueueWorker()
+    {
+        return app('queue.worker');
+    }
+}