added tests and fixed some major bugs

SqlQueue.php

@@ -8,6 +8,8 @@
 use yii\db\Expression;
 use yii\base\Component;
 use Yii;
+use yii\helpers\ArrayHelper;
+use yii\helpers\Json;
 
 class SqlQueue extends Component implements QueueInterface
 {
@@ -21,8 +23,6 @@ class SqlQueue extends Component implements QueueInterface
      */
     public $default = 'default';
 
-    private $_query;
-
     public function init()
     {
         parent::init();
@@ -60,9 +60,9 @@ private function hasTable()
     private function createTable()
     {
         $this->connection->createCommand()->createTable($this->getTableName(), [
-            'id' => 'pk COMMENT "1.0.0"',
+            'id' => 'pk',
             'queue' => 'string(255)',
-            'run_at' => 'timestamp default CURRENT_TIMESTAMP NOT NULL',
+            'run_at' => 'INTEGER NOT NULL',
             'payload' => 'text',
         ])->execute();
         $this->connection->schema->refresh();
@@ -84,29 +84,22 @@ private function getTableName()
     public function push($payload, $queue = null, $delay = 0)
     {
         $this->connection->schema->insert($this->getTableName(), [
-            'queue' => $this->getQueue($queue),
-            'payload' => $payload,
-            'run_at' => new Expression('FROM_UNIXTIME(:unixtime)', [
-                ':unixtime' => time() + $delay,
-            ])
+            'queue' => $queue,
+            'payload' => Json::encode($payload),
+            'run_at' => time() + $delay,
         ]);
         return $this->connection->lastInsertID;
     }
 
     private function getQuery($queue)
     {
-        if ($this->_query) {
-            return $this->_query;
-        }
-
-        $this->_query=new Query;
-        $this->_query->select('id, payload')
-            ->from($this->getTableName())
-            ->where(['queue'=>$queue])
-            ->andWhere('run_at <= NOW()')
+        $query=new Query;
+        $query->from($this->getTableName())
+            ->andFilterWhere(['queue'=>$queue])
+            ->andWhere('run_at <= :timestamp', ['timestamp' => time()])
             ->limit(1);
 
-        return $this->_query;
+        return $query;
     }
 
     /**
@@ -122,9 +115,9 @@ public function delete(array $message)
      */
     public function pop($queue = null)
     {
-        $row=$this->getQuery($this->getQueue($queue))->one($this->connection);
+        $row=$this->getQuery($queue)->one($this->connection);
         if ($row) {
-            return $row['payload'];
+            return ArrayHelper::merge($row, json_decode($row['payload']));
         }
         return false;
     }
@@ -144,7 +137,7 @@ public function release(array $message, $delay = 0)
     {
         $this->connection->createCommand()->update(
             $this->getTableName(),
-            ['run_at' => new Expression('DATE_ADD(NOW(), INTERVAL :delay SECOND)', ['delay' => $delay])],
+            ['run_at' => time() + $delay],
             'id = :id',
             ['id' => $message['id']]
         )->execute();

tests/SqlQueueTest.php

@@ -0,0 +1,101 @@
+<?php
+
+namespace yii\queue\tests;
+
+use yii\db\Connection;
+use yii\db\Query;
+
+class SqlQueueTest extends TestCase
+{
+
+    public function setUp()
+    {
+        $this->mockApplication([
+            'components' => [
+                'queue' => [
+                    'class'=> 'yii\queue\SqlQueue'
+                ],
+                'db' => [
+                    'class' => 'yii\db\Connection',
+                    'dsn' => 'sqlite::memory:',
+                    'charset' => 'utf8',
+                ]
+            ]
+        ]);
+    }
+
+    protected function pushJobToQueue($data = 'test', $delay = 0)
+    {
+        $job=new TestJob([
+            'data'=>$data
+        ]);
+        return $job->push($delay);
+    }
+
+    public function testJobCreation()
+    {
+        $job=new TestJob([
+            'data'=>'test'
+        ]);
+        $this->assertEquals('test', $job->data);
+    }
+
+    public function testJobInsertion()
+    {
+        $result=$this->pushJobToQueue();
+        $resolvedJob=\Yii::$app->queue->pop();
+        $this->assertEquals($result, $resolvedJob['id']);
+    }
+
+
+    public function testJobRun()
+    {
+        $this->pushJobToQueue(__FUNCTION__);
+        $job=\Yii::$app->queue->pop();
+        $jobObject = call_user_func($job['serializer'][1], $job['object']);
+        $this->assertEquals(__FUNCTION__, $jobObject->data);
+        $this->assertEquals(__FUNCTION__, $jobObject->run());
+    }
+
+    public function testJobDelete()
+    {
+        $this->pushJobToQueue(__FUNCTION__);
+        $job=\Yii::$app->queue->pop();
+        \Yii::$app->queue->delete($job);
+
+        $this->assertFalse(\Yii::$app->queue->pop());
+    }
+
+    public function testDifferentQueueWontPopJob()
+    {
+        $this->pushJobToQueue();
+        $this->assertFalse(\Yii::$app->queue->pop('hurrdurrImasheep'));
+    }
+
+    public function testJobPurge()
+    {
+        for ($i = 0; $i <= 10; $i++) {
+            $this->pushJobToQueue(time());
+        }
+        \Yii::$app->queue->purge('test');
+
+        $this->assertFalse(\Yii::$app->queue->pop());
+    }
+
+    public function testJobRelease()
+    {
+        $this->pushJobToQueue(__FUNCTION__);
+        $job = \Yii::$app->queue->pop();
+
+        $this->assertEquals(time(), $job['run_at']);
+
+        \Yii::$app->queue->release($job, 200);
+        $this->assertFalse(\Yii::$app->queue->pop());
+    }
+
+    public function testJobInFutureDoesNotPopNow()
+    {
+        $this->pushJobToQueue(__FUNCTION__, 200);
+        $this->assertFalse(\Yii::$app->queue->pop());
+    }
+}

tests/TestCase.php

@@ -0,0 +1,46 @@
+<?php
+namespace yii\queue\tests;
+
+use yii\helpers\ArrayHelper;
+
+abstract class TestCase extends \PHPUnit_Framework_TestCase
+{
+    /**
+     * Clean up after test.
+     * By default the application created with [[mockApplication]] will be destroyed.
+     */
+    protected function tearDown()
+    {
+        parent::tearDown();
+        $this->destroyApplication();
+    }
+    /**
+     * Populates Yii::$app with a new application
+     * The application will be destroyed on tearDown() automatically.
+     * @param array $config The application configuration, if needed
+     * @param string $appClass name of the application class to create
+     */
+    protected function mockApplication($config = [], $appClass = '\yii\console\Application')
+    {
+        new $appClass(ArrayHelper::merge([
+            'id' => 'testapp',
+            'basePath' => __DIR__,
+            'vendorPath' => $this->getVendorPath(),
+        ], $config));
+    }
+    protected function getVendorPath()
+    {
+        $vendor = dirname(dirname(__DIR__)) . '/vendor';
+        if (!is_dir($vendor)) {
+            $vendor = dirname(dirname(dirname(dirname(__DIR__))));
+        }
+        return $vendor;
+    }
+    /**
+     * Destroys application in Yii::$app by setting it to null.
+     */
+    protected function destroyApplication()
+    {
+        \Yii::$app = null;
+    }
+}

tests/TestJob.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace yii\queue\tests;
+
+use yii\queue\ActiveJob;
+
+class TestJob extends ActiveJob
+{
+    public $data;
+
+    public function queueName()
+    {
+        return 'test';
+    }
+
+    public function run()
+    {
+        return $this->data;
+    }
+}

tests/bootstrap.php

@@ -0,0 +1,11 @@
+<?php
+
+// ensure we get report on all possible php errors
+error_reporting(-1);
+define('YII_ENABLE_ERROR_HANDLER', false);
+define('YII_DEBUG', true);
+$_SERVER['SCRIPT_NAME'] = '/' . __DIR__;
+$_SERVER['SCRIPT_FILENAME'] = __FILE__;
+require_once(__DIR__ . '/../vendor/autoload.php');
+require_once(__DIR__ . '/../vendor/yiisoft/yii2/Yii.php');
+Yii::setAlias('@yii/queue', dirname(__DIR__));