database驱动支持设置connection

Author: yunwuxin

composer.json

@@ -24,7 +24,6 @@
     "minimum-stability": "dev",
     "require": {
         "ext-json": "*",
-        "topthink/think-helper": "^3.0.4",
         "topthink/framework": "^6.0",
         "symfony/process": "^4.2",
         "nesbot/carbon": "^2.16"

src/config.php

@@ -16,9 +16,10 @@
             'type' => 'sync',
         ],
         'database' => [
-            'type'  => 'database',
-            'queue' => 'default',
-            'table' => 'jobs',
+            'type'       => 'database',
+            'queue'      => 'default',
+            'table'      => 'jobs',
+            'connection' => null,
         ],
         'redis'    => [
             'type'       => 'redis',

src/queue/connector/Database.php

@@ -14,6 +14,7 @@
 use Carbon\Carbon;
 use stdClass;
 use think\Db;
+use think\db\ConnectionInterface;
 use think\db\Query;
 use think\queue\Connector;
 use think\queue\InteractsWithTime;
@@ -47,7 +48,7 @@ class Database extends Connector
      */
     protected $retryAfter = 60;
 
-    public function __construct(Db $db, $table, $default = 'default', $retryAfter = 60)
+    public function __construct(ConnectionInterface $db, $table, $default = 'default', $retryAfter = 60)
     {
         $this->db         = $db;
         $this->table      = $table;
@@ -57,7 +58,9 @@ public function __construct(Db $db, $table, $default = 'default', $retryAfter =
 
     public static function __make(Db $db, $config)
     {
-        return new self($db, $config['table'], $config['queue'], $config['retry_after'] ?? 60);
+        $connection = $db->connect($config['connection'] ?? null);
+
+        return new self($connection, $config['table'], $config['queue'], $config['retry_after'] ?? 60);
     }
 
     public function size($queue = null)
@@ -105,9 +108,9 @@ function ($job) use ($queue, $data, $availableAt) {
     /**
      * 重新发布任务
      *
-     * @param string   $queue
+     * @param string $queue
      * @param StdClass $job
-     * @param int      $delay
+     * @param int $delay
      * @return mixed
      */
     public function release($queue, $job, $delay)
@@ -119,9 +122,9 @@ public function release($queue, $job, $delay)
      * Push a raw payload to the database with a given delay.
      *
      * @param \DateTime|int $delay
-     * @param string|null   $queue
-     * @param string        $payload
-     * @param int           $attempts
+     * @param string|null $queue
+     * @param string $payload
+     * @param int $attempts
      * @return mixed
      */
     protected function pushToDatabase($queue, $payload, $delay = 0, $attempts = 0)
@@ -160,7 +163,8 @@ public function pop($queue = null)
     protected function getNextAvailableJob($queue)
     {
 
-        $job = $this->db->name($this->table)
+        $job = $this->db
+            ->name($this->table)
             ->lock(true)
             ->where('queue', $this->getQueue($queue))
             ->where(function (Query $query) {
@@ -190,10 +194,13 @@ protected function getNextAvailableJob($queue)
      */
     protected function markJobAsReserved($job)
     {
-        $this->db->name($this->table)->where('id', $job->id)->update([
-            'reserve_time' => $job->reserve_time = $this->currentTime(),
-            'attempts'     => ++$job->attempts,
-        ]);
+        $this->db
+            ->name($this->table)
+            ->where('id', $job->id)
+            ->update([
+                'reserve_time' => $job->reserve_time = $this->currentTime(),
+                'attempts'     => ++$job->attempts,
+            ]);
 
         return $job;
     }