Use database for queue jobs

Author: barryvdh

src/AsyncQueue.php

@@ -3,6 +3,7 @@
 use Illuminate\Queue\Queue;
 use Illuminate\Queue\QueueInterface;
 use Symfony\Component\Process\Process;
+use Barryvdh\Queue\Models\Job;
 
 class AsyncQueue extends Queue implements QueueInterface {
 
@@ -16,53 +17,54 @@ class AsyncQueue extends Queue implements QueueInterface {
      */
     public function push($job, $data = '', $queue = null)
     {
-        $payload = $this->createPayload($job, $data);
-
-        $process = $this->makeProcess($payload);
-        $process->disableOutput();
-        $process->run();
-
-
+        $id = $this->storeJob($job, $data);
+        $this->startProcess($id, 0);
         return 0;
     }
 
     /**
-     * Create a payload string from the given job and data.
-     *
-     * @param  string $job
-     * @param  mixed  $data
-     * @param  string $queue
-     * @return string
+     * Store the job in the database
+     * 
+     * @param  string  $job
+     * @param  mixed   $data
+     * @param  integer $delay
+     * @return integer The id of the job
      */
-    protected function createPayload($job, $data = '', $queue = null)
-    {
-        $payload = parent::createPayload($job, $data, $queue);
+    public function storeJob($job, $data, $delay = 0){
+
+        $payload = $this->createPayload($job, $data);
+
+        $job = new Job;
+        $job->status = Job::STATUS_OPEN;
+        $job->delay = $delay;
+        $job->payload = $payload;
+        $job->save();
 
-        return base64_encode($payload);
+        return $job->id;
     }
 
     /**
-     * Make a Process for the Artisan command with the payload
+     * Make a Process for the Artisan command for the job id
      *
-     * @param $payload
-     * @return \Symfony\Component\Process\Process
+     * @param  integer $jobId
      */
-    public function makeProcess($payload)
+    public function startProcess($jobId)
     {
         $environment = $this->container->environment();
         $cwd = $this->container['path.base'];
+        $string = 'php artisan queue:async %d --env=%s ';
 
-        $string = 'php artisan queue:async %s --env=%s';
-
-        if (defined('PHP_WINDOWS_VERSION_BUILD')){  
-            $string = 'start /B ' . $string; 
-        } else { 
+        if (defined('PHP_WINDOWS_VERSION_BUILD')){
+            $string = 'start /B ' . $string;
+        } else {
             $string = 'nohup ' . $string . ' &';
-        } 
+        }
 
-        $command = sprintf($string, $payload, $environment);
+        $command = sprintf($string, $jobId, $environment);
 
-        return new Process($command, $cwd);
+        $process = new Process($command, $cwd);
+        $process->disableOutput();
+        $process->run();
     }
 
     /**
@@ -76,7 +78,10 @@ public function makeProcess($payload)
      */
     public function later($delay, $job, $data = '', $queue = null)
     {
-        return $this->push($job, $data, $queue);
+        $delay = $this->getSeconds($delay);
+        $id = $this->storeJob($job, $data, $delay);
+        $this->startProcess($id);
+        return 0;
     }
 
     /**

src/Console/AsyncCommand.php

@@ -1,7 +1,7 @@
 <?php namespace Barryvdh\Queue\Console;
 
+use Barryvdh\Queue\Models\Job;
 use Illuminate\Console\Command;
-use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Input\InputArgument;
 use Barryvdh\Queue\Jobs\AsyncJob;
 
@@ -19,7 +19,7 @@ class AsyncCommand extends Command {
 	 *
 	 * @var string
 	 */
-	protected $description = 'Run a base64+json encode serialized queue';
+	protected $description = 'Run a queue from the database';
 
 	/**
 	 * Create a new command instance.
@@ -37,9 +37,9 @@ public function __construct()
 	 */
 	public function fire()
 	{
-        $payload = $this->argument('payload');
+        $item = Job::findOrFail($this->argument('job_id'));
 
-        $job = new AsyncJob($this->laravel, $payload);
+        $job = new AsyncJob($this->laravel, $item);
 
         $job->fire();
 
@@ -53,18 +53,8 @@ public function fire()
 	protected function getArguments()
 	{
 		return array(
-			array('payload', InputArgument::REQUIRED, 'The Job Payload'),
+			array('job_id', InputArgument::REQUIRED, 'The Job ID'),
 		);
 	}
 
-	/**
-	 * Get the console command options.
-	 *
-	 * @return array
-	 */
-	protected function getOptions()
-	{
-		return array();
-	}
-
-}
+}

src/Jobs/AsyncJob.php

@@ -1,26 +1,27 @@
 <?php namespace Barryvdh\Queue\Jobs;
 
+use Barryvdh\Queue\Models\Job;
 use Illuminate\Queue\Jobs\SyncJob;
 use Illuminate\Container\Container;
 
 class AsyncJob extends SyncJob {
 
     /**
-     * The payload, Base64 & JSON Encoded
+     * The job model
      *
-     * @var array
+     * @var Job
      */
-    protected $payload;
+    protected $job;
 
     /**
      * Create a new job instance.
      *
-     * @param  \Illuminate\Container\Container  $container
-     * @param  string   $payload
+     * @param  \Illuminate\Container\Container $container
+     * @param \Barryvdh\Queue\Models\Job $job
      */
-    public function __construct(Container $container, $payload)
+    public function __construct(Container $container, Job $job)
     {
-        $this->payload = $payload;
+        $this->job = $job;
         $this->container = $container;
     }
 
@@ -31,9 +32,40 @@ public function __construct(Container $container, $payload)
      */
     public function fire()
     {
-        $payload = $this->parsePayload($this->payload);
+        // Get the payload from the job
+        $payload = $this->parsePayload($this->job->payload);
 
+        // If we have to wait, sleep until our time has come
+        if($this->job->delay){
+            $this->job->status = Job::STATUS_WAITING;
+            $this->job->save();
+            sleep($this->job->delay);
+        }
+
+        // Mark job as started
+        $this->job->status = Job::STATUS_STARTED;
+        $this->job->save();
+
+        // Fire the actual job
         $this->resolveAndFire($payload);
+
+        // If job is not deleted, mark as finished
+        if(!$this->deleted){
+            $this->job->status = Job::STATUS_FINISHED;
+            $this->job->save();
+        }
+
+    }
+
+    /**
+     * Delete the job from the queue.
+     *
+     * @return void
+     */
+    public function delete()
+    {
+        parent::delete();
+        $this->job->delete();
     }
 
 
@@ -44,8 +76,8 @@ public function fire()
      * @return array|null
      */
     protected function parsePayload($payload){
-        return json_decode(base64_decode($payload), true);
+        return json_decode($payload, true);
     }
 
 
-}
+}