完善

Author: yunwuxin

composer.json

@@ -23,6 +23,7 @@
     },
     "minimum-stability": "dev",
     "require": {
+        "ext-json": "*",
         "topthink/think-helper": "^3.0.4",
         "topthink/framework": "^6.0.0",
         "topthink/think-factory": "^1.0.0",

src/queue/CallQueuedHandler.php

@@ -11,14 +11,22 @@
 
 namespace think\queue;
 
+use think\App;
+
 class CallQueuedHandler
 {
+    protected $app;
+
+    public function __construct(App $app)
+    {
+        $this->app = $app;
+    }
 
     public function call(Job $job, array $data)
     {
         $command = unserialize($data['command']);
 
-        call_user_func([$command, 'handle']);
+        $this->app->invoke([$command, 'handle']);
 
         if (!$job->isDeletedOrReleased()) {
             $job->delete();

src/queue/Connector.php

@@ -88,23 +88,16 @@ protected function createPlainPayload($job, $data)
 
     protected function createObjectPayload($job)
     {
-        $payload = [
+        return [
             'job'       => 'think\queue\CallQueuedHandler@call',
             'maxTries'  => $job->tries ?? null,
             'timeout'   => $job->timeout ?? null,
             'timeoutAt' => $this->getJobExpiration($job),
             'data'      => [
-                'commandName' => $job,
-                'command'     => $job,
-            ],
-        ];
-
-        return array_merge($payload, [
-            'data' => [
                 'commandName' => get_class($job),
                 'command'     => serialize(clone $job),
             ],
-        ]);
+        ];
     }
 
     public function getJobExpiration($job)

src/queue/Queueable.php

@@ -14,18 +14,33 @@
 trait Queueable
 {
 
+    /** @var string 连接 */
+    public $connection;
+
     /** @var string 队列名称 */
     public $queue;
 
     /** @var integer 延迟时间 */
     public $delay;
 
+    /**
+     * 设置连接名
+     * @param $connection
+     * @return $this
+     */
+    public function onConnection($connection)
+    {
+        $this->connection = $connection;
+
+        return $this;
+    }
+
     /**
      * 设置队列名
      * @param $queue
      * @return $this
      */
-    public function queue($queue)
+    public function onQueue($queue)
     {
         $this->queue = $queue;