优化完善 提升Queue的命名空间

README.md

@@ -4,17 +4,18 @@
 > composer require topthink/think-queue
 
 ## 配置
+> 配置文件位于 `application/extra/queue.php`
 ### 公共配置
 
 ```
-'queue'=>[
-    'type'=>'sync' //驱动类型，可选择 sync(默认):同步执行，database:数据库驱动,redis:Redis驱动,topthink:Topthink驱动
+[
+    'connector'=>'sync' //驱动类型，可选择 sync(默认):同步执行，database:数据库驱动,redis:Redis驱动,topthink:Topthink驱动
                    //或其他自定义的完整的类名
 ]
 ```
 
 ### 驱动配置
-> 各个驱动的具体可用配置项在`think\queue\driver`目录下各个驱动类里的`options`属性中，写在上面的`queue`配置里即可覆盖
+> 各个驱动的具体可用配置项在`think\queue\connector`目录下各个驱动类里的`options`属性中，写在上面的`queue`配置里即可覆盖
 
 
 ## 使用 Database
@@ -108,7 +109,7 @@ class Job2{
 
 
 ## 发布任务
-> `think\queue\Queue:push($job, $data = '', $queue = null)` 和 `think\queue\Queue::later($delay, $job, $data = '', $queue = null)` 两个方法，前者是立即执行，后者是在`$delay`秒后执行
+> `think\Queue:push($job, $data = '', $queue = null)` 和 `think\Queue::later($delay, $job, $data = '', $queue = null)` 两个方法，前者是立即执行，后者是在`$delay`秒后执行
 
 `$job` 是任务名  
 单模块的，且命名空间是`app\job`的，比如上面的例子一,写`Job1`类名即可  

composer.json

@@ -1,20 +1,29 @@
 {
     "name": "topthink/think-queue",
     "description": "The ThinkPHP5 Queue Package",
+    "type": "think-extend",
     "authors": [
         {
             "name": "yunwuxin",
             "email": "[email protected]"
         }
     ],
     "license": "Apache-2.0",
-    "minimum-stability": "dev",
     "autoload": {
         "psr-4": {
-            "think\\queue\\": "src"
+            "think\\": "src"
         },
         "files": [
-            "src/config.php"
+            "src/common.php"
         ]
+    },
+    "require": {
+        "topthink/think-helper": "^1.0",
+        "topthink/think-installer": ">=1.0.10"
+    },
+    "extra": {
+        "think-config": {
+            "queue": "src/config.php"
+        }
     }
 }

src/Queue.php

@@ -9,75 +9,41 @@
 // | Author: yunwuxin <[email protected]>
 // +----------------------------------------------------------------------
 
-namespace think\queue;
-
-use think\Config;
-
+namespace think;
+
+use think\helper\Str;
+use think\queue\Connector;
+
+/**
+ * Class Queue
+ * @package think\queue
+ *
+ * @method static push($job, $data = '', $queue = null)
+ * @method static later($delay, $job, $data = '', $queue = null)
+ * @method static pop($queue = null)
+ * @method static marshal()
+ */
 class Queue
 {
-    protected static $instance = [];
-
-    /**
-     * 添加任务到队列
-     * @param        $job
-     * @param string $data
-     * @param null   $queue
-     */
-    public static function push($job, $data = '', $queue = null)
-    {
-        self::handle()->push($job, $data, $queue);
-
-    }
+    /** @var Connector */
+    protected static $connector;
 
-    /**
-     * 添加延迟任务到队列
-     * @param        $delay
-     * @param        $job
-     * @param string $data
-     * @param null   $queue
-     */
-    public static function later($delay, $job, $data = '', $queue = null)
-    {
-        self::handle()->later($delay, $job, $data, $queue);
-    }
-
-    /**
-     * 获取第一个任务
-     * @param null $queue
-     * @return mixed
-     */
-    public static function pop($queue = null)
-    {
-        if (!method_exists(self::handle(), 'pop'))
-            throw new \RuntimeException('pop queues not support for this type');
-
-        return self::handle()->pop($queue);
-    }
-
-
-    /**
-     * 由订阅的推送执行任务
-     */
-    public static function marshal()
-    {
-        if (!method_exists(self::handle(), 'marshal'))
-            throw new \RuntimeException('push queues not support for this type');
-
-        self::handle()->marshal();
-    }
-
-    private static function handle()
+    private static function buildConnector()
     {
         $options = Config::get('queue');
         $type    = !empty($options['type']) ? $options['type'] : 'Sync';
 
-        if (!isset(self::$instance[$type])) {
+        if (!isset(self::$connector)) {
 
-            $class = false !== strpos($type, '\\') ? $type : '\\think\\queue\\driver\\' . ucwords($type);
+            $class = false !== strpos($type, '\\') ? $type : '\\think\\queue\\connector\\' . Str::studly($type);
 
-            self::$instance[$type] = new $class($options);
+            self::$connector = new $class($options);
         }
-        return self::$instance[$type];
+        return self::$connector;
     }
 
+    public static function __callStatic($name, $arguments)
+    {
+        return call_user_func_array([self::buildConnector(), $name], $arguments);
+    }
 }

src/common.php

@@ -0,0 +1,17 @@
+<?php
+// +----------------------------------------------------------------------
+// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
+// +----------------------------------------------------------------------
+// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
+// +----------------------------------------------------------------------
+// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
+// +----------------------------------------------------------------------
+// | Author: yunwuxin <[email protected]>
+// +----------------------------------------------------------------------
+
+\think\Console::addDefaultCommands([
+    "think\\queue\\command\\Work",
+    "think\\queue\\command\\Restart",
+    "think\\queue\\command\\Listen",
+    "think\\queue\\command\\Subscribe"
+]);

src/config.php

@@ -2,16 +2,13 @@
 // +----------------------------------------------------------------------
 // | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
 // +----------------------------------------------------------------------
-// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
+// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.
 // +----------------------------------------------------------------------
 // | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
 // +----------------------------------------------------------------------
 // | Author: yunwuxin <[email protected]>
 // +----------------------------------------------------------------------
 
-\think\Console::addDefaultCommands([
-    "think\\queue\\command\\Work",
-    "think\\queue\\command\\Restart",
-    "think\\queue\\command\\Listen",
-    "think\\queue\\command\\Subscribe"
-]);
+return [
+    'connector' => 'Sync'
+];

src/queue/CallQueuedHandler.php

@@ -0,0 +1,36 @@
+<?php
+// +----------------------------------------------------------------------
+// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
+// +----------------------------------------------------------------------
+// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.
+// +----------------------------------------------------------------------
+// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
+// +----------------------------------------------------------------------
+// | Author: yunwuxin <[email protected]>
+// +----------------------------------------------------------------------
+
+namespace think\queue;
+
+class CallQueuedHandler
+{
+
+    public function call(Job $job, array $data)
+    {
+        $command = unserialize($data['command']);
+
+        call_user_func([$command, 'handle']);
+
+        if (!$job->isDeletedOrReleased()) {
+            $job->delete();
+        }
+    }
+
+    public function failed(array $data, $e)
+    {
+        $command = unserialize($data['command']);
+
+        if (method_exists($command, 'failed')) {
+            $command->failed($e);
+        }
+    }
+}

src/queue/Connector.php

@@ -0,0 +1,69 @@
+<?php
+// +----------------------------------------------------------------------
+// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
+// +----------------------------------------------------------------------
+// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.
+// +----------------------------------------------------------------------
+// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
+// +----------------------------------------------------------------------
+// | Author: yunwuxin <[email protected]>
+// +----------------------------------------------------------------------
+
+namespace think\queue;
+
+use InvalidArgumentException;
+
+abstract class Connector
+{
+    protected $options = [];
+
+    abstract function push($job, $data = '', $queue = null);
+
+    abstract function later($delay, $job, $data = '', $queue = null);
+
+    abstract public function pop($queue = null);
+
+    public function marshal()
+    {
+        throw new \RuntimeException('pop queues not support for this type');
+    }
+
+    protected function createPayload($job, $data = '', $queue = null)
+    {
+        if (is_object($job)) {
+            $payload = json_encode([
+                'job'  => 'think\queue\CallQueuedHandler@call',
+                'data' => [
+                    'commandName' => get_class($job),
+                    'command'     => serialize(clone $job),
+                ],
+            ]);
+        } else {
+            $payload = json_encode($this->createPlainPayload($job, $data));
+        }
+
+        if (JSON_ERROR_NONE !== json_last_error()) {
+            throw new InvalidArgumentException('Unable to create payload: ' . json_last_error_msg());
+        }
+
+        return $payload;
+    }
+
+    protected function createPlainPayload($job, $data)
+    {
+        return ['job' => $job, 'data' => $data];
+    }
+
+    protected function setMeta($payload, $key, $value)
+    {
+        $payload       = json_decode($payload, true);
+        $payload[$key] = $value;
+        $payload       = json_encode($payload);
+
+        if (JSON_ERROR_NONE !== json_last_error()) {
+            throw new InvalidArgumentException('Unable to create payload: ' . json_last_error_msg());
+        }
+
+        return $payload;
+    }
+}

src/driver/Database.php → src/queue/connector/Database.php

@@ -9,12 +9,13 @@
 // | Author: yunwuxin <[email protected]>
 // +----------------------------------------------------------------------
 
-namespace think\queue\driver;
+namespace think\queue\connector;
 
 use think\Db;
+use think\queue\Connector;
 use think\queue\job\Database as DatabaseJob;
 
-class Database
+class Database extends Connector
 {
     protected $db;
 
@@ -44,12 +45,6 @@ public function later($delay, $job, $data = '', $queue = null)
         return $this->pushToDatabase($delay, $queue, $this->createPayload($job, $data));
     }
 
-
-    protected function createPayload($job, $data)
-    {
-        return json_encode(['job' => $job, 'data' => $data]);
-    }
-
     public function pop($queue = null)
     {
         $queue = $this->getQueue($queue);