feat: add support for leaf redis

Author: mychidarko

src/Queue/Adapters/Redis.php

@@ -16,9 +16,8 @@ class Redis implements Adapter
 
     protected array $config = [];
 
-    public function __construct($config = [])
+    public function __construct()
     {
-        $this->config = $config;
         $this->redis = new \Leaf\Redis();
     }
 
@@ -27,11 +26,19 @@ public function __construct($config = [])
      */
     public function connect($connection)
     {
-        $this->redis->init($connection);
+        $this->config['table'] = $connection['table'] ?? 'leaf_php_jobs';
+
+        if (redis()->ping()) {
+            $this->redis = redis();
+        } else {
+            $this->redis->connect(MvcConfig('redis'));
+        }
 
         if (!$this->redis->get($this->config['table'])) {
             $this->redis->set($this->config['table'], json_encode([]));
         }
+
+        return $this;
     }
 
     /**
@@ -84,6 +91,7 @@ public function setJobStatus($id, $status)
         foreach ($jobs as $key => $job) {
             if ($job['id'] === $id) {
                 $jobs[$key]['status'] = $status;
+
                 $this->redis->set($this->config['table'], json_encode($jobs));
 
                 return true;
@@ -116,28 +124,25 @@ public function getJobs()
      */
     public function getNextJob()
     {
-        $jobs = $this->redis->get($this->config['table']) ?? [];
-        $jobs = json_decode($jobs, true);
-
-        $job = array_values(array_filter($jobs, function ($job) {
-            return $job['status'] === 'pending';
-        }))[0] ?? null;
+        foreach ($this->getJobs() as $job) {
+            if ($job['status'] === 'pending') {
+                return $job;
+            }
+        }
 
-        return $job;
+        return null;
     }
 
     /**
      * @inheritDoc
      */
     public function retryFailedJob($id, $retryCount)
     {
-        $jobs = $this->redis->get($this->config['table']) ?? [];
-        $jobs = json_decode($jobs, true);
-
-        foreach ($jobs as $key => $job) {
+        foreach ($this->getJobs() as $key => $job) {
             if ($job['id'] === $id) {
                 $jobs[$key]['status'] = 'pending';
                 $jobs[$key]['retry_count'] = (int) $retryCount + 1;
+
                 $this->redis->set($this->config['table'], json_encode($jobs));
 
                 return true;

src/Queue/Commands/QueueWorkCommand.php

@@ -6,18 +6,42 @@
 
 class QueueWorkCommand extends Command
 {
+    protected $queueConfig;
+
     protected static $defaultName = 'queue:work';
     public $description = 'Start your queue worker';
     public $help = 'Start your queue worker';
 
+    protected function config()
+    {
+        $this->queueConfig = MvcConfig('queue');
+        $this->setOption('queue', 'queue', 'optional', 'The queue you want to run', $this->queueConfig['default']);
+    }
+
     protected function handle()
     {
-        $this->writeln('Queue worker started for queue \'default\'...');
+        $queue = $this->option('queue');
+
+        $this->writeln("Queue worker started for queue '$queue'...");
+
+        if ($this->queueConfig['connections'][$queue]['driver'] === 'database') {
+            $this->writeln('> Using database connection for queue...');
+
+            if (!file_exists(DatabasePath("{$this->queueConfig['connections'][$queue]['table']}.yml"))) {
+                $this->writeln("> Queue table not found. Creating queue table...");
 
-        $queueConfig = MvcConfig('queue');
+                \Leaf\FS\File::copy(__DIR__ . '/stubs/schema.yml', DatabasePath("{$this->queueConfig['connections'][$queue]['table']}.yml"));
+                \Aloe\Core::run(
+                    "php leaf db:migrate {$this->queueConfig['connections'][$queue]['table']}",
+                    $this->output
+                );
+            }
+        } else {
+            $this->writeln('> Using redis connection for queue...');
+        }
 
         (new \Leaf\Worker())
-            ->queue($queueConfig['connections'][$queueConfig['default']])
+            ->queue($this->queueConfig['connections'][$queue])
             ->run();
 
         return 0;