Initial commit

Author: ilzrv

.gitignore

@@ -0,0 +1,3 @@
+.idea/
+/vendor/
+.DS_Store

composer.json

@@ -0,0 +1,22 @@
+{
+    "name": "ilzrv/php-bull-queue",
+    "description": "PHP Job Creator For Bull Queue",
+    "require": {
+        "php": "^7.4",
+        "ramsey/uuid": "^4.1"
+    },
+    "license": "MIT",
+    "authors": [
+        {
+            "name": "Ilia Lazarev",
+            "email": "[email protected]"
+        }
+    ],
+    "autoload": {
+        "psr-4": {
+            "Ilzrv\\PhpBullQueue\\": "src"
+        }
+    }
+}
+
+

src/DTOs/DataTransferObject.php

@@ -0,0 +1,42 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ilzrv\PhpBullQueue\DTOs;
+
+use ReflectionClass;
+use ReflectionNamedType;
+use ReflectionProperty;
+
+abstract class DataTransferObject
+{
+    public function __construct(array $parameters = [])
+    {
+        $class = new ReflectionClass(static::class);
+
+        foreach ($class->getProperties(ReflectionProperty::IS_PUBLIC) as $reflectionProperty) {
+            $property = $reflectionProperty->getName();
+
+            if (isset($parameters[$property])) {
+                $this->{$property} = $parameters[$property];
+            } else {
+                /** @var ReflectionNamedType $type */
+                $type = $reflectionProperty->getType();
+                $name = $type->getName();
+
+                $this->{$property} = !$type->isBuiltin() && class_exists($name)
+                    ? new $name()
+                    : $this->{$property} ?? null;
+            }
+
+//            $this->{$property} = !$type->isBuiltin() && class_exists($name)
+//                ? $parameters[$property] ?? new $name()
+//                : $parameters[$property] ?? $this->{$property} ?? null;
+        }
+    }
+
+    public function toJson(): string
+    {
+        return (string) json_encode($this);
+    }
+}

src/DTOs/JobOpts.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ilzrv\PhpBullQueue\DTOs;
+
+class JobOpts extends DataTransferObject
+{
+    public string $customJobId = '';
+    public int $priority = 0;
+    public bool $lifo = false;
+    public int $attempts = 1;
+    public int $timestamp;
+    public int $delay = 0;
+}

src/DTOs/QueueOpts.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ilzrv\PhpBullQueue\DTOs;
+
+class QueueOpts extends DataTransferObject
+{
+    public RedisConfig $redis;
+    public string $prefix = 'bull';
+}

src/DTOs/RedisConfig.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ilzrv\PhpBullQueue\DTOs;
+
+class RedisConfig extends DataTransferObject
+{
+    public string $driver = 'phpredis';
+    public string $host = '127.0.0.1';
+    public int $port = 6379;
+}

src/Drivers/PhpRedisQueue.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ilzrv\PhpBullQueue\Drivers;
+
+use Ilzrv\PhpBullQueue\DTOs\RedisConfig;
+use Redis;
+
+class PhpRedisQueue implements RedisQueue
+{
+    protected Redis $client;
+
+    public function __construct(RedisConfig $config, Redis $redis = null)
+    {
+        if (!is_null($redis)) {
+            $this->client = $redis;
+        } else {
+            $this->client = new Redis();
+
+            $this->client->connect(
+                $config->host,
+                $config->port,
+            );
+        }
+    }
+
+    public function add(string $script, array $args, int $numKeys)
+    {
+        return $this->client->eval($script, $args, $numKeys);
+    }
+}

src/Drivers/PredisQueue.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ilzrv\PhpBullQueue\Drivers;
+
+use Ilzrv\PhpBullQueue\DTOs\RedisConfig;
+use Predis\Client as Redis;
+
+class PredisQueue implements RedisQueue
+{
+    protected Redis $client;
+
+    public function __construct(RedisConfig $config, Redis $redis = null)
+    {
+        if (!is_null($redis)) {
+            $this->client = $redis;
+        } else {
+            $this->client = new Redis([
+                'host' => $config->host,
+                'port' => $config->port,
+            ]);
+        }
+    }
+
+    public function add(string $script, array $args, int $numKeys)
+    {
+        return $this->client->eval($script, $numKeys, ...$args);
+    }
+}

src/Drivers/RedisQueue.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ilzrv\PhpBullQueue\Drivers;
+
+interface RedisQueue
+{
+    public function add(string $script, array $args, int $numKeys);
+}

src/LuaScripts.php

@@ -0,0 +1,103 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ilzrv\PhpBullQueue;
+
+class LuaScripts
+{
+    /**
+     * Adds a job to the queue by doing the following:
+     *
+     * KEYS[1] 'wait',
+     * KEYS[2] 'paused'
+     * KEYS[3] 'meta-paused'
+     * KEYS[4] 'id'
+     * KEYS[5] 'delayed'
+     * KEYS[6] 'priority'
+     *
+     * ARGV[1] key prefix
+     * ARGV[2] custom id (will not generate one automatically)
+     * ARGV[3] name
+     * ARGV[4] data (json stringified job data)
+     * ARGV[5] opts (json stringified job opts)
+     * ARGV[6] timestamp
+     * ARGV[7] delay
+     * ARGV[8] delayedTimestamp
+     * ARGV[9] priority
+     * ARGV[10] LIFO
+     * ARGV[11] token
+     *
+     * @return string
+     */
+    public static function add()
+    {
+        return <<<'LUA'
+local jobId
+local jobIdKey
+local rcall = redis.call
+
+local jobCounter = rcall("INCR", KEYS[4])
+
+if ARGV[2] == "" then
+  jobId = jobCounter
+  jobIdKey = ARGV[1] .. jobId
+else
+  jobId = ARGV[2]
+  jobIdKey = ARGV[1] .. jobId
+  if rcall("EXISTS", jobIdKey) == 1 then
+    return jobId .. "" -- convert to string
+  end
+end
+
+-- Store the job.
+rcall("HMSET", jobIdKey, "name", ARGV[3], "data", ARGV[4], "opts", ARGV[5], "timestamp", ARGV[6], "delay", ARGV[7], "priority", ARGV[9])
+
+-- Check if job is delayed
+local delayedTimestamp = tonumber(ARGV[8])
+if(delayedTimestamp ~= 0) then
+  local timestamp = delayedTimestamp * 0x1000 + bit.band(jobCounter, 0xfff)
+  rcall("ZADD", KEYS[5], timestamp, jobId)
+  rcall("PUBLISH", KEYS[5], delayedTimestamp)
+else
+  local target
+
+  -- Whe check for the meta-paused key to decide if we are paused or not
+  -- (since an empty list and !EXISTS are not really the same)
+  local paused
+  if rcall("EXISTS", KEYS[3]) ~= 1 then
+    target = KEYS[1]
+    paused = false
+  else
+    target = KEYS[2]
+    paused = true
+  end
+
+  -- Standard or priority add
+  local priority = tonumber(ARGV[9])
+  if priority == 0 then
+      -- LIFO or FIFO
+    rcall(ARGV[10], target, jobId)
+  else
+    -- Priority add
+    rcall("ZADD", KEYS[6], priority, jobId)
+    local count = rcall("ZCOUNT", KEYS[6], 0, priority)
+
+    local len = rcall("LLEN", target)
+    local id = rcall("LINDEX", target, len - (count-1))
+    if id then
+      rcall("LINSERT", target, "BEFORE", id, jobId)
+    else
+      rcall("RPUSH", target, jobId)
+    end
+
+  end
+
+  -- Emit waiting event (wait..ing@token)
+  rcall("PUBLISH", KEYS[1] .. "ing@" .. ARGV[11], jobId)
+end
+
+return jobId .. "" -- convert to string
+LUA;
+    }
+}

src/Queue.php

@@ -0,0 +1,84 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ilzrv\PhpBullQueue;
+
+use Ilzrv\PhpBullQueue\Drivers\PhpRedisQueue;
+use Ilzrv\PhpBullQueue\Drivers\PredisQueue;
+use Ilzrv\PhpBullQueue\Drivers\RedisQueue;
+use Ilzrv\PhpBullQueue\DTOs\JobOpts;
+use Ilzrv\PhpBullQueue\DTOs\QueueOpts;
+use Ramsey\Uuid\Uuid;
+use RuntimeException;
+
+class Queue
+{
+    const DEFAULT_JOB_NAME = '__default__';
+
+    private string $name;
+
+    private QueueOpts $opts;
+
+    public function __construct(string $name, QueueOpts $opts = null)
+    {
+        $this->name = $name;
+        $this->opts = $opts ?: new QueueOpts();
+    }
+
+    /**
+     * Adds a job to the queue.
+     *
+     * @param string $name
+     * @param array $data
+     * @param JobOpts|null $opts
+     * @return mixed
+     */
+    public function add(string $name, array $data, JobOpts $opts = null)
+    {
+        $opts = $opts ?: new JobOpts([
+            'timestamp' => (int) str_replace('.', '', microtime(true)),
+        ]);
+
+        $prefix = sprintf('%s:%s:', $this->opts->prefix, $this->name);
+
+        return $this->queue()->add(
+            LuaScripts::add(),
+            [
+                $prefix.'wait',
+                $prefix.'paused',
+                $prefix.'meta-paused',
+                $prefix.'id',
+                $prefix.'delayed',
+                $prefix.'priority',
+                $prefix,
+                $opts->customJobId,
+                $name,
+                json_encode($data),
+                $opts->toJson(),
+                $opts->timestamp,
+                $opts->delay,
+                $opts->delay ? $opts->timestamp + $opts->delay : 0,
+                $opts->priority,
+                $opts->lifo ? 'RPUSH' : 'LPUSH',
+                (string) Uuid::uuid4()
+            ],
+            6
+        );
+    }
+
+    /**
+     * @return RedisQueue
+     */
+    protected function queue()
+    {
+        switch ($this->opts->redis->driver) {
+            case 'predis':
+                return new PredisQueue($this->opts->redis);
+            case 'phpredis':
+                return new PhpRedisQueue($this->opts->redis);
+            default:
+                throw new RuntimeException("{$this->opts->redis->driver} driver not supported.");
+        }
+    }
+}