[8.x] Add queue:monitor command (#38168)

* add queue:monitor command

* fix style

* formatting

Co-authored-by: Taylor Otwell <[email protected]>

Author: themsaid

src/Illuminate/Foundation/Providers/ArtisanServiceProvider.php

@@ -68,6 +68,7 @@
 use Illuminate\Queue\Console\ForgetFailedCommand as ForgetFailedQueueCommand;
 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\PruneBatchesCommand as PruneBatchesQueueCommand;
 use Illuminate\Queue\Console\PruneFailedJobsCommand;
 use Illuminate\Queue\Console\RestartCommand as QueueRestartCommand;
@@ -111,6 +112,7 @@ class ArtisanServiceProvider extends ServiceProvider implements DeferrableProvid
         'QueueFlush' => 'command.queue.flush',
         'QueueForget' => 'command.queue.forget',
         'QueueListen' => 'command.queue.listen',
+        'QueueMonitor' => 'command.queue.monitor',
         'QueuePruneBatches' => 'command.queue.prune-batches',
         'QueuePruneFailedJobs' => 'command.queue.prune-failed-jobs',
         'QueueRestart' => 'command.queue.restart',
@@ -702,6 +704,18 @@ protected function registerQueueListenCommand()
         });
     }
 
+    /**
+     * Register the command.
+     *
+     * @return void
+     */
+    protected function registerQueueMonitorCommand()
+    {
+        $this->app->singleton('command.queue.monitor', function ($app) {
+            return new QueueMonitorCommand($app['queue'], $app['events']);
+        });
+    }
+
     /**
      * Register the command.
      *

src/Illuminate/Queue/Console/MonitorCommand.php

@@ -0,0 +1,137 @@
+<?php
+
+namespace Illuminate\Queue\Console;
+
+use Illuminate\Console\Command;
+use Illuminate\Contracts\Events\Dispatcher;
+use Illuminate\Contracts\Queue\Factory;
+use Illuminate\Queue\Events\QueueBusy;
+use Illuminate\Support\Collection;
+
+class MonitorCommand extends Command
+{
+    /**
+     * The console command name.
+     *
+     * @var string
+     */
+    protected $signature = 'queue:monitor
+                       {queues : The names of the queues to monitor}
+                       {--max=1000 : The maximum number of jobs that can be on the queue before an event is dispatched}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Monitor the size of the specified queues';
+
+    /**
+     * The queue manager instance.
+     *
+     * @var \Illuminate\Contracts\Queue\Factory
+     */
+    protected $manager;
+
+    /**
+     * The events dispatcher instance.
+     *
+     * @var \Illuminate\Contracts\Events\Dispatcher
+     */
+    protected $events;
+
+    /**
+     * The table headers for the command.
+     *
+     * @var string[]
+     */
+    protected $headers = ['Connection', 'Queue', 'Size', 'Status'];
+
+    /**
+     * Create a new queue listen command.
+     *
+     * @param  \Illuminate\Contracts\Queue\Factory  $manager
+     * @param  \Illuminate\Contracts\Events\Dispatcher  $events
+     * @return void
+     */
+    public function __construct(Factory $manager, Dispatcher $events)
+    {
+        parent::__construct();
+
+        $this->manager = $manager;
+        $this->events = $events;
+    }
+
+    /**
+     * Execute the console command.
+     *
+     * @return void
+     */
+    public function handle()
+    {
+        $queues = $this->parseQueues($this->argument('queues'));
+
+        $this->displaySizes($queues);
+
+        $this->dispatchEvents($queues);
+    }
+
+    /**
+     * Parse the queues into an array of the connections and queues.
+     *
+     * @param  string  $queues
+     * @return \Illuminate\Support\Collection
+     */
+    protected function parseQueues($queues)
+    {
+        return collect(explode(',', $queues))->map(function ($queue) {
+            [$connection, $queue] = array_pad(explode(':', $queue, 2), 2, null);
+
+            if (! isset($queue)) {
+                $queue = $connection;
+                $connection = $this->laravel['config']['queue.default'];
+            }
+
+            return [
+                'connection' => $connection,
+                'queue' => $queue,
+                'size' => $size = $this->manager->connection($connection)->size($queue),
+                'status' => $size >= $this->option('max') ? '<fg=red>ALERT</>' : 'OK',
+            ];
+        });
+    }
+
+    /**
+     * Display the failed jobs in the console.
+     *
+     * @param  \Illuminate\Support\Collection  $queues
+     * @return void
+     */
+    protected function displaySizes(Collection $queues)
+    {
+        $this->table($this->headers, $queues);
+    }
+
+    /**
+     * Fire the monitoring events.
+     *
+     * @param  \Illuminate\Support\Collection  $queues
+     * @return void
+     */
+    protected function dispatchEvents(Collection $queues)
+    {
+        foreach ($queues as $queue) {
+            if ($queue['status'] == 'OK') {
+                continue;
+            }
+
+            $this->events->dispatch(
+                new QueueBusy(
+                    $queue['connection'],
+                    $queue['queue'],
+                    $queue['size'],
+                )
+            );
+        }
+    }
+}

src/Illuminate/Queue/Events/QueueBusy.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace Illuminate\Queue\Events;
+
+class QueueBusy
+{
+    /**
+     * The connection name.
+     *
+     * @var string
+     */
+    public $connection;
+
+    /**
+     * The queue name.
+     *
+     * @var string
+     */
+    public $queue;
+
+    /**
+     * The size of the queue.
+     *
+     * @var int
+     */
+    public $size;
+
+    /**
+     * Create a new event instance.
+     *
+     * @param  string $connection
+     * @param  string $queue
+     * @param  int  $size
+     * @return void
+     */
+    public function __construct($connection, $queue, $size)
+    {
+        $this->connection = $connection;
+        $this->queue = $queue;
+        $this->size = $size;
+    }
+}