Add blocking mode with auto-reconnection and some other features.

.gitignore

@@ -6,3 +6,4 @@ composer.lock
 phpunit.xml
 .phpunit.result.cache
 .php_cs.cache
+.phpunit.cache

README.md

@@ -557,6 +557,50 @@ There are two ways of consuming messages.
 
 2. `rabbitmq:consume` command which is provided by this package. This command utilizes `basic_consume` and is more performant than `basic_get` by ~2x, but does not support multiple queues.
 
+### Getting performance and durability
+To get the best performance and durability you must use option 2, and you need to extend the base config and use some extra options in the command.
+
+The command to execute:
+```bash
+  rabbitmq:consume --blocking=1 --auto-reconnect=1 --alive-check=30 --init-queue=1 --verbose-messages=1 --prefetch-count=1 
+```
+- `--blocking=1` is used to use blocking waiting mechanism [performance] (STRONGLY RECOMMENDED)
+- `--auto-reconnect=1` is used to auto reconnect if something is wrong with the connection [durability] (recommended)
+- `--alive-check=30` is used to custom check if connection is stuck [durability] (recommended with `blocking=1` if you have virtualization/proxies)
+- `--init-queue=1` is used to auto create queue before consuming [durability] (recommended)
+- `--verbose-messages=1` is used to not write anything about processed messages [performance] (not necessary)
+- `--prefetch-count=1` is used to limit number of messages prefetched [durability] (not necessary)
+
+
+The config to provide:
+```php
+'connections' => [
+    // ...
+
+    'rabbitmq' => [
+        // ...
+
+        'options' => [
+            // ...            
+            'queue' => [
+                'use_expiration_for_delayed_queues' => true // To create only one later-queue for any TTL
+                'declare_full_route' => true // To auto-init full route of message if you need both queue + exchange
+                'retries' => [ // To enable retries if the queue fails to push
+                    'enabled' => true,
+                    'max' => 5, // number of retries
+                    'pause_micro_seconds' => 1e6 // pause between retries
+                ],
+
+                'channel_rpc_timeout' => 3 // To make custom alive-check work (required with `blocking=1` and --alive-check=N`)
+                'keepalive' => true // To keep the connection alive (STRONGLY RECOMMENDED with `blocking=1`)
+            ]
+        ],
+    ],
+
+    // ...    
+],
+```
+
 ## Testing
 
 Setup RabbitMQ using `docker-compose`:

src/Console/ConsumeCommand.php

@@ -3,6 +3,10 @@
 namespace VladimirYuldashev\LaravelQueueRabbitMQ\Console;
 
 use Illuminate\Queue\Console\WorkCommand;
+use Illuminate\Queue\Events\JobFailed;
+use Illuminate\Queue\Events\JobProcessed;
+use Illuminate\Queue\Events\JobProcessing;
+use Illuminate\Queue\Events\JobReleasedAfterException;
 use Illuminate\Support\Str;
 use VladimirYuldashev\LaravelQueueRabbitMQ\Consumer;
 
@@ -30,10 +34,18 @@ class ConsumeCommand extends WorkCommand
                             {--consumer-tag}
                             {--prefetch-size=0}
                             {--prefetch-count=1000}
+                            {--blocking=0 : Weather to block queue waiting or not}
+                            {--init-queue=0 : Enables init the queue before starting consuming}
+                            {--auto-reconnect=0 : Enables auto-reconnection when something is wrong with the connection}
+                            {--auto-reconnect-pause=0.5 : The pause (in seconds) before reconnecting}
+                            {--alive-check=0 : The pause (in seconds) before reconnecting}
+                            {--verbose-messages=0 : Write messages only when verbose mode is enabled}
                            ';
 
     protected $description = 'Consume messages';
 
+    protected $useVerboseForMessages = false;
+
     public function handle(): void
     {
         /** @var Consumer $consumer */
@@ -45,6 +57,14 @@ public function handle(): void
         $consumer->setMaxPriority((int) $this->option('max-priority'));
         $consumer->setPrefetchSize((int) $this->option('prefetch-size'));
         $consumer->setPrefetchCount((int) $this->option('prefetch-count'));
+        $consumer->setBlocking($this->booleanOption('blocking'));
+        $consumer->setInitQueue($this->booleanOption('init-queue'));
+        $consumer->setAutoReconnect($this->booleanOption('auto-reconnect'));
+        $consumer->setAutoReconnectPause((float)$this->option('auto-reconnect-pause'));
+        $consumer->setAliveCheck((float)$this->option('alive-check'));
+
+        $consumer->setInteractWithIO($this);
+        $this->useVerboseForMessages = $this->booleanOption('verbose-messages');
 
         parent::handle();
     }
@@ -63,4 +83,45 @@ protected function consumerTag(): string
 
         return Str::substr($consumerTag, 0, 255);
     }
+
+    protected function booleanOption(string $key): bool
+    {
+        return filter_var(
+            $this->option($key),
+            FILTER_VALIDATE_BOOLEAN
+        );
+    }
+
+    /**
+     * Output worker results only in verbose mode
+     */
+    protected function listenForEvents()
+    {
+        if ($this->useVerboseForMessages) {
+            parent::listenForEvents();
+            return;
+        }
+
+        $this->laravel['events']->listen(JobFailed::class, function ($event) {
+            if ($this->output->isVerbose()) {
+                $this->writeOutput($event->job, 'failed');
+            }
+
+            $this->logFailedJob($event);
+        });
+
+        if ($this->output->isVerbose()) {
+            $this->laravel['events']->listen(JobProcessing::class, function ($event) {
+                $this->writeOutput($event->job, 'starting');
+            });
+
+            $this->laravel['events']->listen(JobProcessed::class, function ($event) {
+                $this->writeOutput($event->job, 'success');
+            });
+
+            $this->laravel['events']->listen(JobReleasedAfterException::class, function ($event) {
+                $this->writeOutput($event->job, 'released_after_exception');
+            });
+        }
+    }
 }