[13.x] Add queue concurrency driver

config/concurrency.php

@@ -11,10 +11,45 @@
     | by Laravel's concurrency functions. By default, concurrent work will
     | be sent to isolated PHP processes which will return their results.
     |
-    | Supported: "process", "fork", "sync"
+    | Supported: "process", "fork", "sync", "queue"
     |
     */
 
     'default' => env('CONCURRENCY_DRIVER', 'process'),
 
+    /*
+    |--------------------------------------------------------------------------
+    | Concurrency Drivers
+    |--------------------------------------------------------------------------
+    |
+    | Below you may configure each of the concurrency drivers utilized by your
+    | application. The queue driver will distribute your tasks to the queue
+    | workers, which return their results through the given cache store.
+    |
+    */
+
+    'drivers' => [
+
+        'process' => [
+            'driver' => 'process',
+        ],
+
+        'fork' => [
+            'driver' => 'fork',
+        ],
+
+        'sync' => [
+            'driver' => 'sync',
+        ],
+
+        'queue' => [
+            'driver' => 'queue',
+            'connection' => env('CONCURRENCY_QUEUE_CONNECTION'),
+            'queue' => env('CONCURRENCY_QUEUE'),
+            'store' => env('CONCURRENCY_CACHE_STORE'),
+            'timeout' => (int) env('CONCURRENCY_TIMEOUT', 60),
+        ],
+
+    ],
+
 ];

src/Illuminate/Concurrency/CapturedTaskException.php

@@ -0,0 +1,19 @@
+<?php
+
+namespace Illuminate\Concurrency;
+
+use RuntimeException;
+use Throwable;
+
+/**
+ * The original task exception has already been captured in the task's
+ * result envelope for the caller. This wrapper is rethrown so that the
+ * failure remains visible to the queue's failed job machinery.
+ */
+class CapturedTaskException extends RuntimeException
+{
+    public function __construct(Throwable $previous)
+    {
+        parent::__construct($previous->getMessage(), 0, $previous);
+    }
+}

src/Illuminate/Concurrency/ConcurrencyManager.php

@@ -2,7 +2,13 @@
 
 namespace Illuminate\Concurrency;
 
+use Illuminate\Bus\Queueable;
+use Illuminate\Cache\CacheManager;
+use Illuminate\Contracts\Bus\Dispatcher;
+use Illuminate\Contracts\Cache\Factory as CacheFactory;
 use Illuminate\Process\Factory as ProcessFactory;
+use Illuminate\Queue\QueueManager;
+use Illuminate\Support\Arr;
 use Illuminate\Support\MultipleInstanceManager;
 use RuntimeException;
 use Spatie\Fork\Fork;
@@ -28,21 +34,23 @@ public function driver($name = null)
     /**
      * Create an instance of the process concurrency driver.
      *
+     * @param  array  $config
      * @return \Illuminate\Concurrency\ProcessDriver
      */
-    public function createProcessDriver()
+    public function createProcessDriver(array $config = [])
     {
         return new ProcessDriver($this->app->make(ProcessFactory::class));
     }
 
     /**
      * Create an instance of the fork concurrency driver.
      *
+     * @param  array  $config
      * @return \Illuminate\Concurrency\ForkDriver
      *
      * @throws \RuntimeException
      */
-    public function createForkDriver()
+    public function createForkDriver(array $config = [])
     {
         if (! $this->app->runningInConsole()) {
             throw new RuntimeException('Due to PHP limitations, the fork driver may not be used within web requests.');
@@ -58,13 +66,38 @@ public function createForkDriver()
     /**
      * Create an instance of the sync concurrency driver.
      *
+     * @param  array  $config
      * @return \Illuminate\Concurrency\SyncDriver
      */
-    public function createSyncDriver()
+    public function createSyncDriver(array $config = [])
     {
         return new SyncDriver;
     }
 
+    /**
+     * Create an instance of the queue concurrency driver.
+     *
+     * @param  array  $config
+     * @return \Illuminate\Concurrency\QueueDriver
+     *
+     * @throws \RuntimeException
+     */
+    public function createQueueDriver(array $config = [])
+    {
+        if (! trait_exists(Queueable::class) ||
+            ! class_exists(QueueManager::class) ||
+            ! class_exists(CacheManager::class)) {
+            throw new RuntimeException('Please install the "illuminate/bus", "illuminate/cache", and "illuminate/queue" Composer packages in order to utilize the "queue" driver.');
+        }
+
+        return new QueueDriver(
+            $this->app->make(Dispatcher::class),
+            $this->app->make(CacheFactory::class),
+            $this->app->make('config'),
+            Arr::except($config, ['driver']),
+        );
+    }
+
     /**
      * Get the default instance name.
      *
@@ -97,6 +130,12 @@ public function setDefaultInstance($name)
      */
     public function getInstanceConfig($name)
     {
+        $config = $this->app['config']->get('concurrency.drivers.'.$name);
+
+        if (is_array($config)) {
+            return array_merge(['driver' => $name], $config);
+        }
+
         return $this->app['config']->get(
             'concurrency.driver.'.$name, ['driver' => $name],
         );

src/Illuminate/Concurrency/Console/InvokeSerializedClosureCommand.php

@@ -2,8 +2,8 @@
 
 namespace Illuminate\Concurrency\Console;
 
+use Illuminate\Concurrency\TaskResult;
 use Illuminate\Console\Command;
-use ReflectionClass;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Throwable;
 
@@ -41,41 +41,17 @@ class InvokeSerializedClosureCommand extends Command
     public function handle()
     {
         try {
-            $this->output->write(json_encode([
-                'successful' => true,
-                'result' => serialize($this->laravel->call(match (true) {
-                    ! is_null($this->argument('code')) => unserialize($this->argument('code')),
-                    isset($_SERVER['LARAVEL_INVOKABLE_CLOSURE']) => unserialize(
-                        base64_decode($_SERVER['LARAVEL_INVOKABLE_CLOSURE'])
-                    ),
-                    default => fn () => null,
-                })),
-            ]));
+            $this->output->write(json_encode(TaskResult::success($this->laravel->call(match (true) {
+                ! is_null($this->argument('code')) => unserialize($this->argument('code')),
+                isset($_SERVER['LARAVEL_INVOKABLE_CLOSURE']) => unserialize(
+                    base64_decode($_SERVER['LARAVEL_INVOKABLE_CLOSURE'])
+                ),
+                default => fn () => null,
+            }))));
         } catch (Throwable $e) {
             report($e);
 
-            $reflection = new ReflectionClass($e);
-            $constructor = $reflection->getConstructor();
-            $parameters = [];
-
-            if ($constructor) {
-                $declaringClass = $constructor->getDeclaringClass()->getName();
-
-                if ($declaringClass === $reflection->getName()) {
-                    foreach ($constructor->getParameters() as $parameter) {
-                        $parameters[$parameter->name] = $e->{$parameter->name} ?? null;
-                    }
-                }
-            }
-
-            $this->output->write(json_encode([
-                'successful' => false,
-                'exception' => get_class($e),
-                'message' => $e->getMessage(),
-                'file' => $e->getFile(),
-                'line' => $e->getLine(),
-                'parameters' => $parameters,
-            ]));
+            $this->output->write(json_encode(TaskResult::failure($e)));
         }
     }
 }

src/Illuminate/Concurrency/InvokeDeferredClosure.php

@@ -0,0 +1,50 @@
+<?php
+
+namespace Illuminate\Concurrency;
+
+use Closure;
+use Illuminate\Contracts\Container\Container;
+use Illuminate\Queue\CallQueuedClosure;
+use Illuminate\Queue\Jobs\SyncJob;
+use Laravel\SerializableClosure\SerializableClosure;
+use Throwable;
+
+/**
+ * A queued closure that reports a failure on a sync connection instead of
+ * rethrowing it, so a failover connection does not treat the failure as the
+ * connection being down and run the task again.
+ */
+class InvokeDeferredClosure extends CallQueuedClosure
+{
+    /**
+     * Create a new job instance.
+     *
+     * @param  \Closure  $job
+     * @return static
+     */
+    public static function create(Closure $job)
+    {
+        return new static(new SerializableClosure($job));
+    }
+
+    /**
+     * Execute the job.
+     *
+     * @param  \Illuminate\Contracts\Container\Container  $container
+     * @return void
+     */
+    public function handle(Container $container)
+    {
+        try {
+            parent::handle($container);
+        } catch (Throwable $e) {
+            if ($this->job instanceof SyncJob) {
+                report($e);
+
+                return;
+            }
+
+            throw $e;
+        }
+    }
+}