update queue, worker, job reliance

Author: mychidarko

src/Job.php

@@ -2,7 +2,9 @@
 
 namespace Leaf;
 
-class Job
+use Leaf\Queue\Dispatchable;
+
+abstract class Job implements Dispatchable
 {
     /**
      * Job config
@@ -14,20 +16,46 @@ class Job
      */
     protected $job = [];
 
+    /**
+     * Data to pass to the job
+     */
+    protected $data = [];
+
     /**
      * Queue instance
      * @var \Leaf\Queue
      */
     protected $queue = null;
 
+    /**
+     * Configured queue connection
+     */
+    protected string $connection = 'default';
+
+    /**
+     * Return configured connection
+     */
+    public function connection()
+    {
+        return $this->connection;
+    }
+
     /**
      * Load a job for running
      */
-    public function __construct($job, $config, $queue)
+    public function fromQueue($job, $config, $queue)
     {
         $this->job = $job;
-        $this->config = $config;
         $this->queue = $queue;
+
+        if (isset($config['data'])) {
+            $this->data = $config['data'];
+            unset($config['data']);
+        }
+
+        $this->config = $config;
+
+        return $this;
     }
 
     /**
@@ -73,15 +101,6 @@ public function hasReachedRetryLimit()
         return $this->job['retry_count'] >= $this->config['tries'];
     }
 
-    /**
-     * Handle job retry limit
-     */
-    public function handleRetryLimit()
-    {
-        echo "Job {$this->job['id']} has reached retry limit\n";
-        $this->queue->pop($this->job['id']);
-    }
-
     /**
      * Set job status
      */
@@ -96,17 +115,19 @@ public function setStatus($status)
     public function retry()
     {
         sleep($this->config['delayBeforeRetry'] ?? 0);
-        $this->queue->retryFailedJob($this->job['id'], $this->config['retry_count']);
+        $this->queue->retryFailedJob($this->job['id'], $this->job['retry_count']);
     }
 
     /**
      * Release the job back into the queue after (n) seconds.
      *
-     * @param  int  $delay
+     * @param int  $delay
      * @return void
      */
     public function release($delay = 0)
     {
+        sleep($delay);
+
         $this->queue->push([
             'class' => $this->job['class'],
             'config' => json_encode($this->config),
@@ -116,46 +137,39 @@ public function release($delay = 0)
     }
 
     /**
-     * Create a new job
+     * Pass data to the job
      */
-    public static function create(callable $job)
+    public function with($data)
     {
-        //
+        $this->data = $data;
+        return $this;
     }
 
-    public function handle()
+    public function stack()
     {
-        //
+        return;
     }
 
-    public static function dispatch($config = [], $queue = 'default')
+    public function getConfig()
     {
-        $queue = new Queue();
-        $queueConfig = MvcConfig('queue') ?? [];
-
-        if (empty($queueConfig)) {
-            if (!file_exists($configFile = \Aloe\Command\Config::rootpath('queue.config.php'))) {
-                throw new \Exception('Queue config not found');
-            }
-
-            $queueConfig = require $configFile;
-        }
-
-        $queue->config($queueConfig);
-        $queue->connect();
-
-        return $queue->push([
-            'class' => get_called_class(),
-            'config' => json_encode(array_merge($queue->config()['workerConfig'] ?? [], $config)),
-            'status' => 'pending',
-            'retry_count' => 0,
-        ]);
+        return [
+            'delay' => 0,
+            'delayBeforeRetry' => 0,
+            'expire' => 60,
+            'force' => false,
+            'memory' => 128,
+            'quitOnEmpty' => false,
+            'sleep' => 3,
+            'timeout' => 60,
+            'tries' => 3,
+            'data' => $this->data,
+        ];
     }
 
     public function trigger()
     {
         $this->queue->setJobStatus($this->job['id'], 'processing');
-        $this->handle();
+        $this->handle(...$this->data);
     }
 
     public function removeFromQueue()

src/Queue.php

@@ -15,57 +15,19 @@ class Queue
     /**
      * The queue config
      */
-    protected $config = [
-        'archive' => false,
-        'adapter' => 'db',
-        'connection' => [],
-        'table' => 'leafphp_queue_main',
-        'workers' => 1,
-        'workerConfig' => [
-            'delay' => 0,
-            'delayBeforeRetry' => 0,
-            'expire' => 60,
-            'force' => false,
-            'memory' => 128,
-            'quitOnEmpty' => false,
-            'sleep' => 3,
-            'timeout' => 60,
-            'tries' => 3,
-        ],
-    ];
-
-    /**
-     * Config for the queue and worker
-     * @param array $config The config for the queue and worker
-     */
-    public function config(?array $config = null)
-    {
-        if (!$config) {
-            return $this->config;
-        }
-
-        $this->config = array_merge(
-            $this->config,
-            \Leaf\Config::get('queue') ?? [],
-            $config ?? [],
-        );
-
-        return $this->config;
-    }
+    protected $connection = [];
 
     /**
      * Connect to queue adapter
+     * @param array $connection The connection to use
      * @return \Leaf\Queue
      */
-    public function connect(): Queue
+    public function connect($connection): Queue
     {
-        $config = $this->config(\Leaf\Config::get('queue') ?? []);
-
-        $adapter = ucfirst($config['adapter']);
+        $adapter = ucfirst($connection['driver']);
         $adapter = "\\Leaf\\Queue\\Adapters\\$adapter";
 
-        $this->adapter = new $adapter($config);
-        $this->adapter->connect($config['connections'][$config['default'] ?? 'redis']);
+        $this->adapter = (new $adapter())->connect($connection);
 
         return $this;
     }
@@ -84,10 +46,6 @@ public function getAdapter()
      */
     public function push(array $job)
     {
-        if (!$this->adapter) {
-            $this->connect();
-        }
-
         $this->adapter->pushJobToQueue($job);
     }
 
@@ -97,10 +55,6 @@ public function push(array $job)
      */
     public function pop($id)
     {
-        if (!$this->adapter) {
-            $this->connect();
-        }
-
         $this->adapter->popJobFromQueue($id);
     }
 
@@ -112,10 +66,6 @@ public function pop($id)
      */
     public function setJobStatus($id, $status)
     {
-        if (!$this->adapter) {
-            $this->connect();
-        }
-
         $this->adapter->setJobStatus($id, $status);
     }
 
@@ -125,10 +75,6 @@ public function setJobStatus($id, $status)
      */
     public function getNextJob()
     {
-        if (!$this->adapter) {
-            $this->connect();
-        }
-
         return $this->adapter->getNextJob();
     }
 
@@ -138,10 +84,6 @@ public function getNextJob()
      */
     public function markJobAsFailed($id)
     {
-        if (!$this->adapter) {
-            $this->connect();
-        }
-
         $this->adapter->markJobAsFailed($id);
     }
 
@@ -152,10 +94,6 @@ public function markJobAsFailed($id)
      */
     public function retryFailedJob($id, $retryCount = 0)
     {
-        if (!$this->adapter) {
-            $this->connect();
-        }
-
         $this->adapter->retryFailedJob($id, $retryCount);
     }
 
@@ -167,24 +105,7 @@ public static function commands()
         return [
             \Leaf\Queue\Commands\DeleteJobCommand::class,
             \Leaf\Queue\Commands\GenerateJobCommand::class,
-            \Leaf\Queue\Commands\QueueConfigCommand::class,
-            \Leaf\Queue\Commands\QueueInstallCommand::class,
-            \Leaf\Queue\Commands\QueuePauseCommand::class,
-            \Leaf\Queue\Commands\QueueRunCommand::class,
+            \Leaf\Queue\Commands\QueueWorkCommand::class,
         ];
     }
-
-    /**
-     * Initialize a worker to work the queue
-     */
-    public function run()
-    {
-        for ($i = 0; $i < $this->config['workers']; $i++) {
-            $worker = new Worker();
-            $worker
-                ->config($this->config['workerConfig'])
-                ->queue($this)
-                ->run();
-        }
-    }
 }

src/Queue/Adapters/Adapter.php

@@ -9,8 +9,6 @@
  */
 interface Adapter
 {
-    public function __construct($config = []);
-
     /**
      * Connect to queue storage
      * @param array $connection Credentials for the queue storage