Feature/queue pause resume

src/Illuminate/Foundation/Providers/ArtisanServiceProvider.php

@@ -97,9 +97,12 @@
 use Illuminate\Queue\Console\ListenCommand as QueueListenCommand;
 use Illuminate\Queue\Console\ListFailedCommand as ListFailedQueueCommand;
 use Illuminate\Queue\Console\MonitorCommand as QueueMonitorCommand;
+use Illuminate\Queue\Console\PauseCommand as QueuePauseCommand;
+use Illuminate\Queue\Console\PauseListCommand as QueuePauseListCommand;
 use Illuminate\Queue\Console\PruneBatchesCommand as QueuePruneBatchesCommand;
 use Illuminate\Queue\Console\PruneFailedJobsCommand as QueuePruneFailedJobsCommand;
 use Illuminate\Queue\Console\RestartCommand as QueueRestartCommand;
+use Illuminate\Queue\Console\ResumeCommand as QueueResumeCommand;
 use Illuminate\Queue\Console\RetryBatchCommand as QueueRetryBatchCommand;
 use Illuminate\Queue\Console\RetryCommand as QueueRetryCommand;
 use Illuminate\Queue\Console\TableCommand;
@@ -150,9 +153,12 @@ class ArtisanServiceProvider extends ServiceProvider implements DeferrableProvid
         'QueueForget' => ForgetFailedQueueCommand::class,
         'QueueListen' => QueueListenCommand::class,
         'QueueMonitor' => QueueMonitorCommand::class,
+        'QueuePause' => QueuePauseCommand::class,
+        'QueuePauseList' => QueuePauseListCommand::class,
         'QueuePruneBatches' => QueuePruneBatchesCommand::class,
         'QueuePruneFailedJobs' => QueuePruneFailedJobsCommand::class,
         'QueueRestart' => QueueRestartCommand::class,
+        'QueueResume' => QueueResumeCommand::class,
         'QueueRetry' => QueueRetryCommand::class,
         'QueueRetryBatch' => QueueRetryBatchCommand::class,
         'QueueWork' => QueueWorkCommand::class,

src/Illuminate/Queue/Console/PauseCommand.php

@@ -0,0 +1,62 @@
+<?php
+
+namespace Illuminate\Queue\Console;
+
+use Illuminate\Console\Command;
+use Illuminate\Contracts\Queue\Factory as QueueManager;
+use Symfony\Component\Console\Attribute\AsCommand;
+
+#[AsCommand(name: 'queue:pause')]
+class PauseCommand extends Command
+{
+    /**
+     * The console command name.
+     *
+     * @var string
+     */
+    protected $signature = 'queue:pause {queue : The name of the queue to pause}
+                                       {--ttl=86400 : The TTL for the pause in seconds}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Pause processing for a specific queue';
+
+    /**
+     * The queue manager instance.
+     *
+     * @var \Illuminate\Contracts\Queue\Factory
+     */
+    protected $manager;
+
+    /**
+     * Create a new queue pause command.
+     *
+     * @param  \Illuminate\Contracts\Queue\Factory  $manager
+     */
+    public function __construct(QueueManager $manager)
+    {
+        parent::__construct();
+
+        $this->manager = $manager;
+    }
+
+    /**
+     * Execute the console command.
+     *
+     * @return int
+     */
+    public function handle()
+    {
+        $queue = $this->argument('queue');
+        $ttl = (int) $this->option('ttl');
+
+        $this->manager->pause($queue, $ttl);
+
+        $this->components->info("Queue [{$queue}] has been paused.");
+
+        return 0;
+    }
+}

src/Illuminate/Queue/Console/PauseListCommand.php

@@ -0,0 +1,69 @@
+<?php
+
+namespace Illuminate\Queue\Console;
+
+use Illuminate\Console\Command;
+use Illuminate\Contracts\Queue\Factory as QueueManager;
+use Symfony\Component\Console\Attribute\AsCommand;
+
+#[AsCommand(name: 'queue:pause:list')]
+class PauseListCommand extends Command
+{
+    /**
+     * The console command name.
+     *
+     * @var string
+     */
+    protected $name = 'queue:pause:list';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'List all currently paused queues';
+
+    /**
+     * The queue manager instance.
+     *
+     * @var \Illuminate\Contracts\Queue\Factory
+     */
+    protected $manager;
+
+    /**
+     * Create a new queue pause list command.
+     *
+     * @param  \Illuminate\Contracts\Queue\Factory  $manager
+     */
+    public function __construct(QueueManager $manager)
+    {
+        parent::__construct();
+
+        $this->manager = $manager;
+    }
+
+    /**
+     * Execute the console command.
+     *
+     * @return int
+     */
+    public function handle()
+    {
+        $pausedQueues = $this->manager->getPausedQueues();
+
+        if (empty($pausedQueues)) {
+            $this->components->info('No queues are currently paused.');
+
+            return 0;
+        }
+
+        $this->components->info('Paused Queues:');
+
+        $this->table(
+            ['Queue Name'],
+            collect($pausedQueues)->map(fn ($queue) => [$queue])->toArray()
+        );
+
+        return 0;
+    }
+}

src/Illuminate/Queue/Console/ResumeCommand.php

@@ -0,0 +1,66 @@
+<?php
+
+namespace Illuminate\Queue\Console;
+
+use Illuminate\Console\Command;
+use Illuminate\Contracts\Queue\Factory as QueueManager;
+use Symfony\Component\Console\Attribute\AsCommand;
+
+#[AsCommand(name: 'queue:resume')]
+class ResumeCommand extends Command
+{
+    /**
+     * The console command name.
+     *
+     * @var string
+     */
+    protected $signature = 'queue:resume {queue : The name of the queue to resume}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Resume processing for a paused queue';
+
+    /**
+     * The queue manager instance.
+     *
+     * @var \Illuminate\Contracts\Queue\Factory
+     */
+    protected $manager;
+
+    /**
+     * Create a new queue resume command.
+     *
+     * @param  \Illuminate\Contracts\Queue\Factory  $manager
+     */
+    public function __construct(QueueManager $manager)
+    {
+        parent::__construct();
+
+        $this->manager = $manager;
+    }
+
+    /**
+     * Execute the console command.
+     *
+     * @return int
+     */
+    public function handle()
+    {
+        $queue = $this->argument('queue');
+
+        if (! $this->manager->isPaused($queue)) {
+            $this->components->warn("Queue [{$queue}] is not paused.");
+
+            return 1;
+        }
+
+        $this->manager->resume($queue);
+
+        $this->components->info("Queue [{$queue}] has been resumed.");
+
+        return 0;
+    }
+}

src/Illuminate/Queue/QueueManager.php

@@ -295,6 +295,88 @@ public function setApplication($app)
         return $this;
     }
 
+    /**
+     * Pause a queue by name.
+     *
+     * @param  string  $queue
+     * @param  int  $ttl
+     * @return void
+     */
+    public function pause($queue, $ttl = 86400)
+    {
+        $cache = $this->app['cache']->store();
+
+        $cache->put("queue_paused:{$queue}", true, $ttl);
+
+        // Add to the list of paused queues
+        $pausedQueues = $this->getPausedQueues();
+
+        if (! in_array($queue, $pausedQueues)) {
+            $pausedQueues[] = $queue;
+            $this->storePausedQueuesList($pausedQueues);
+        }
+    }
+
+    /**
+     * Resume a paused queue by name.
+     *
+     * @param  string  $queue
+     * @return void
+     */
+    public function resume($queue)
+    {
+        $cache = $this->app['cache']->store();
+
+        $cache->forget("queue_paused:{$queue}");
+
+        // Remove from the list of paused queues
+        $pausedQueues = $this->getPausedQueues();
+
+        if (($key = array_search($queue, $pausedQueues)) !== false) {
+            unset($pausedQueues[$key]);
+            $this->storePausedQueuesList(array_values($pausedQueues));
+        }
+    }
+
+    /**
+     * Determine if a queue is paused.
+     *
+     * @param  string  $queue
+     * @return bool
+     */
+    public function isPaused($queue)
+    {
+        $cache = $this->app['cache']->store();
+
+        return (bool) $cache->get("queue_paused:{$queue}", false);
+    }
+
+    /**
+     * Get all paused queues.
+     *
+     * @return array
+     */
+    public function getPausedQueues()
+    {
+        $cache = $this->app['cache']->store();
+
+        // This implementation stores a list of paused queues
+        return $cache->get('queue_paused_list', []);
+    }
+
+    /**
+     * Store the list of paused queues.
+     *
+     * @param  array  $queues
+     * @return void
+     */
+    protected function storePausedQueuesList($queues)
+    {
+        $cache = $this->app['cache']->store();
+
+        $cache->forever('queue_paused_list', $queues);
+    }
+
     /**
      * Dynamically pass calls to the default connection.
      *

src/Illuminate/Queue/Worker.php

@@ -372,6 +372,11 @@ protected function getNextJob($connection, $queue)
             }
 
             foreach (explode(',', $queue) as $index => $queue) {
+                // Skip paused queues
+                if ($this->isQueuePaused($queue)) {
+                    continue;
+                }
+
                 if (! is_null($job = $popJobCallback($queue, $index))) {
                     $this->raiseAfterJobPopEvent($connection->getConnectionName(), $job);
 
@@ -387,6 +392,21 @@ protected function getNextJob($connection, $queue)
         }
     }
 
+    /**
+     * Determine if a queue is paused.
+     *
+     * @param  string  $queue
+     * @return bool
+     */
+    protected function isQueuePaused($queue)
+    {
+        if (! $this->cache) {
+            return false;
+        }
+
+        return (bool) $this->cache->get("queue_paused:{$queue}", false);
+    }
+
     /**
      * Process the given job.
      *

src/Illuminate/Support/Facades/Queue.php

@@ -62,6 +62,10 @@
  * @method static array pushedJobs()
  * @method static array rawPushes()
  * @method static \Illuminate\Support\Testing\Fakes\QueueFake serializeAndRestore(bool $serializeAndRestore = true)
+ * @method static void pause(string $queue, int $ttl = 86400)
+ * @method static void resume(string $queue)
+ * @method static bool isPaused(string $queue)
+ * @method static array getPausedQueues()
  *
  * @see \Illuminate\Queue\QueueManager
  * @see \Illuminate\Queue\Queue