[5.7] Transactional jobs

src/Illuminate/Contracts/Queue/Transactional.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace Illuminate\Contracts\Queue;
+
+interface Transactional
+{
+}

src/Illuminate/Queue/Jobs/Job.php

@@ -3,6 +3,7 @@
 namespace Illuminate\Queue\Jobs;
 
 use Illuminate\Support\InteractsWithTime;
+use Illuminate\Contracts\Queue\Transactional;
 
 abstract class Job
 {
@@ -80,7 +81,23 @@ public function fire()
 
         [$class, $method] = JobName::parse($payload['job']);
 
-        ($this->instance = $this->resolve($class))->{$method}($this, $payload['data']);
+        $this->instance = $this->resolve($class);
+
+        $handler = function () use ($method, $payload) {
+            $this->instance->{$method}($this, $payload['data']);
+        };
+
+        if ($this->instance instanceof Transactional) {
+            $handler = function () use ($handler) {
+                $connection = method_exists($this->instance, 'dbConnection') && \is_callable([$this->instance, 'dbConnection'])
+                    ? $this->instance->dbConnection()
+                    : null;
+
+                $this->container->make('db')->getConnection($connection)->transaction($handler);
+            };
+        }
+
+        $handler();
     }
 
     /**

tests/Queue/QueueSyncTransactionalJobTest.php

@@ -0,0 +1,92 @@
+<?php
+
+namespace Illuminate\Tests\Queue;
+
+use Illuminate\Queue\SyncQueue;
+use PHPUnit\Framework\TestCase;
+use Illuminate\Queue\Jobs\SyncJob;
+use Illuminate\Container\Container;
+use Illuminate\Database\Connection;
+use Illuminate\Database\Capsule\Manager;
+use Illuminate\Contracts\Queue\Transactional;
+
+class QueueTransactionalJobTest extends TestCase
+{
+    /**
+     * @param $jobClassname
+     * @param $isTransactional
+     * @throws \ReflectionException
+     *
+     * @dataProvider jobsDataProvider
+     */
+    public function testRunFireInDatabaseTransaction($jobClassname, $isTransactional, $connectionName)
+    {
+        $connection = $this->createMock(Connection::class);
+        $connection->expects($isTransactional ? $this->once() : $this->never())
+            ->method('transaction')
+            ->willReturnCallback(function ($handler) {
+                $handler();
+            });
+
+        $db = $this->createMock(Manager::class);
+        $db->expects($isTransactional ? $this->once() : $this->never())
+            ->method('getConnection')
+            ->with($connectionName)
+            ->willReturn($connection);
+
+        /** @var Container|\PHPUnit_Framework_MockObject_MockObject $container */
+        $container = $this->createMock(Container::class);
+        $container->expects($this->exactly($isTransactional ? 2 : 1))
+            ->method('make')
+            ->willReturnCallback(function ($abstract) use ($db) {
+                switch ($abstract) {
+                    case 'db':
+                        return $db;
+                    default:
+                        return new $abstract;
+                }
+            });
+
+        $payloadFactory = new \ReflectionMethod(SyncQueue::class, 'createPayload');
+        $payloadFactory->setAccessible(true);
+
+        $payload = $payloadFactory->invoke(new SyncQueue(), $jobClassname, null, '');
+
+        unset($_SERVER['0(*_*)0']);
+        $job = new SyncJob($container, $payload, '', '');
+        $job->fire();
+        $this->assertEquals('0(*_*)0', $_SERVER['0(*_*)0']);
+    }
+
+    /**
+     * @return array
+     */
+    public function jobsDataProvider(): array
+    {
+        return [
+            [TransactionalWithCustomConnection::class, true, 'custom'],
+            [TransactionalJob::class, true, null],
+            [SimpleJob::class, false, null],
+        ];
+    }
+}
+
+class SimpleJob
+{
+    public function fire()
+    {
+        $_SERVER['0(*_*)0'] = '0(*_*)0';
+    }
+}
+
+class TransactionalJob extends SimpleJob implements Transactional
+{
+}
+
+class TransactionalWithCustomConnection extends TransactionalJob
+{
+    public function dbConnection(): string
+    {
+        return 'custom';
+    }
+}