5.5 database queue - transactions wrapped in closures, retry attempts

- fixes #7046
laravel · Dec 12, 2017 · 46046e7 · 46046e7
1 parent 242d438
commit 46046e7
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 17 deletions.
diff --git a/src/Illuminate/Queue/Connectors/DatabaseConnector.php b/src/Illuminate/Queue/Connectors/DatabaseConnector.php
@@ -37,7 +37,8 @@ public function connect(array $config)
             $this->connections->connection($config['connection'] ?? null),
             $config['table'],
             $config['queue'],
-            $config['retry_after'] ?? 60
+            $config['retry_after'] ?? 60,
+            $config['transaction_attempts'] ?? 1
         );
     }
 }
diff --git a/src/Illuminate/Queue/DatabaseQueue.php b/src/Illuminate/Queue/DatabaseQueue.php
@@ -38,6 +38,13 @@ class DatabaseQueue extends Queue implements QueueContract
      */
     protected $retryAfter = 60;
 
+    /**
+     * The number of attempts for DB transaction for job select & job delete.
+     *
+     * @var int
+     */
+    protected $transactionAttempts = 1;
+
     /**
      * Create a new database queue instance.
      *
@@ -47,12 +54,13 @@ class DatabaseQueue extends Queue implements QueueContract
      * @param  int  $retryAfter
      * @return void
      */
-    public function __construct(Connection $database, $table, $default = 'default', $retryAfter = 60)
+    public function __construct(Connection $database, $table, $default = 'default', $retryAfter = 60, $transactionAttempts = 1)
     {
         $this->table = $table;
         $this->default = $default;
         $this->database = $database;
         $this->retryAfter = $retryAfter;
+        $this->transactionAttempts = $transactionAttempts;
     }
 
     /**
@@ -184,18 +192,19 @@ protected function buildDatabaseRecord($queue, $payload, $availableAt, $attempts
      *
      * @param  string  $queue
      * @return \Illuminate\Contracts\Queue\Job|null
+     * @throws \Exception|\Throwable
      */
     public function pop($queue = null)
     {
         $queue = $this->getQueue($queue);
 
-        $this->database->beginTransaction();
-
-        if ($job = $this->getNextAvailableJob($queue)) {
-            return $this->marshalJob($queue, $job);
-        }
+        return $this->database->transaction(function () use ($queue) {
+            if ($job = $this->getNextAvailableJob($queue)) {
+                return $this->marshalJob($queue, $job);
+            }
 
-        $this->database->commit();
+            return null;
+        }, $this->transactionAttempts);
     }
 
     /**
@@ -259,8 +268,6 @@ protected function marshalJob($queue, $job)
     {
         $job = $this->markJobAsReserved($job);
 
-        $this->database->commit();
-
         return new DatabaseJob(
             $this->container, $this, $job, $this->connectionName, $queue
         );
@@ -288,16 +295,15 @@ protected function markJobAsReserved($job)
      * @param  string  $queue
      * @param  string  $id
      * @return void
+     * @throws \Exception|\Throwable
      */
     public function deleteReserved($queue, $id)
     {
-        $this->database->beginTransaction();
-
-        if ($this->database->table($this->table)->lockForUpdate()->find($id)) {
-            $this->database->table($this->table)->where('id', $id)->delete();
-        }
-
-        $this->database->commit();
+        $this->database->transaction(function () use ($queue, $id) {
+            if ($this->database->table($this->table)->lockForUpdate()->find($id)) {
+                $this->database->table($this->table)->where('id', $id)->delete();
+            }
+        }, $this->transactionAttempts);
     }
 
     /**