feat: Make RabbitMQ basic_qos settings configurable

This commit introduces configurable basic_qos settings for the RabbitMQ consumer, allowing users to fine-tune message prefetching behavior.

Key changes:

-   Moved basic_qos settings (prefetch_size, prefetch_count, global) to the configuration file.
-   Users can now adjust these settings to optimize performance based on their specific needs.
-   Updated documentation to reflect the new configuration options.

This change provides greater flexibility and control over message consumption, improving overall performance and resource utilization.

Author: iamfarhad

README.md

@@ -62,6 +62,11 @@ Add connection to config/queue.php:
                 ],
                 'queue'       => [
                     'job' => \iamfarhad\LaravelRabbitMQ\Jobs\RabbitMQJob::class,
+                    'qos' => [
+                        'prefetch_size' => 0,
+                        'prefetch_count' => 10,
+                        'global' => false
+                    ]
                 ],
             ],
         ]

config/RabbitMQConnectionConfig.php

@@ -2,29 +2,34 @@
 
 return [
     'driver' => 'rabbitmq',
-    'queue'  => env('RABBITMQ_QUEUE', 'default'),
+    'queue' => env('RABBITMQ_QUEUE', 'default'),
 
     'hosts' => [
-        'host'      => env('RABBITMQ_HOST', '127.0.0.1'),
-        'port'      => env('RABBITMQ_PORT', 5672),
-        'user'      => env('RABBITMQ_USER', 'guest'),
-        'password'  => env('RABBITMQ_PASSWORD', 'guest'),
-        'vhost'     => env('RABBITMQ_VHOST', '/'),
-        'lazy'      => env('RABBITMQ_LAZY_CONNECTION', true),
+        'host' => env('RABBITMQ_HOST', '127.0.0.1'),
+        'port' => env('RABBITMQ_PORT', 5672),
+        'user' => env('RABBITMQ_USER', 'guest'),
+        'password' => env('RABBITMQ_PASSWORD', 'guest'),
+        'vhost' => env('RABBITMQ_VHOST', '/'),
+        'lazy' => env('RABBITMQ_LAZY_CONNECTION', true),
         'keepalive' => env('RABBITMQ_KEEPALIVE_CONNECTION', false),
         'heartbeat' => env('RABBITMQ_HEARTBEAT_CONNECTION', 0),
     ],
 
     'options' => [
         'ssl_options' => [
-            'cafile'      => env('RABBITMQ_SSL_CAFILE', null),
-            'local_cert'  => env('RABBITMQ_SSL_LOCALCERT', null),
-            'local_key'   => env('RABBITMQ_SSL_LOCALKEY', null),
+            'cafile' => env('RABBITMQ_SSL_CAFILE', null),
+            'local_cert' => env('RABBITMQ_SSL_LOCALCERT', null),
+            'local_key' => env('RABBITMQ_SSL_LOCALKEY', null),
             'verify_peer' => env('RABBITMQ_SSL_VERIFY_PEER', true),
-            'passphrase'  => env('RABBITMQ_SSL_PASSPHRASE', null),
+            'passphrase' => env('RABBITMQ_SSL_PASSPHRASE', null),
         ],
-        'queue'       => [
+        'queue' => [
             'job' => \iamfarhad\LaravelRabbitMQ\Jobs\RabbitMQJob::class,
+            'qos' => [
+                'prefetch_size' => 0,
+                'prefetch_count' => 10,
+                'global' => false
+            ]
         ],
     ],
 ];

src/Console/ConsumeCommand.php

@@ -35,8 +35,6 @@ final class ConsumeCommand extends WorkCommand
                             {--rest=0 : Number of seconds to rest between jobs}
                             {--max-priority=null : Maximum priority level to consume}
                             {--consumer-tag}
-                            {--prefetch-size=0}
-                            {--prefetch-count=1000}
                             {--num-processes=2 : Number of processes to run in parallel}
                            ';
 
@@ -137,10 +135,6 @@ private function consume(): int
             $consumer->setName($this->option('name'));
             $consumer->setConsumerTag($this->generateConsumerTag());
 
-            // Initialize prefetch size and count first
-            $consumer->setPrefetchSize((int) $this->option('prefetch-size'));
-            $consumer->setPrefetchCount((int) $this->option('prefetch-count'));
-
             // Only set max priority if it's provided and not null
             $maxPriority = $this->option('max-priority');
             if ($maxPriority !== null && $maxPriority !== '') {

src/Consumer.php

@@ -17,12 +17,8 @@ class Consumer extends Worker
 
     private string $consumerTag;
 
-    private int $prefetchSize;
-
     private int $maxPriority;
 
-    private int $prefetchCount;
-
     private AMQPChannel $amqpChannel;
 
     private object|null $currentJob = null;
@@ -42,16 +38,6 @@ public function setMaxPriority(int $value): void
         $this->maxPriority = $value;
     }
 
-    public function setPrefetchSize(int $value): void
-    {
-        $this->prefetchSize = $value;
-    }
-
-    public function setPrefetchCount(int $value): void
-    {
-        $this->prefetchCount = $value;
-    }
-
     /**
      * Listen to the given queue in a loop.
      *
@@ -75,12 +61,6 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
 
         $this->amqpChannel = $connection->getChannel();
 
-        $this->amqpChannel->basic_qos(
-            $this->prefetchSize,
-            $this->prefetchCount,
-            null
-        );
-
         $jobClass = $connection->getJobClass();
         $arguments = [];
         if ($this->maxPriority !== 0) {
@@ -90,6 +70,16 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
             ];
         }
 
+        $prefetchSize = config('queue.connections.rabbitmq.queue.prefetch_size') ?? 0;
+        $prefetchCount = config('queue.connections.rabbitmq.queue.prefetch_count') ?? 10;
+        $global = config('queue.connections.rabbitmq.queue.global') ?? false;
+
+        $this->amqpChannel->basic_qos(
+            $prefetchSize,
+            $prefetchCount,
+            $global
+        );
+
         $this->amqpChannel->basic_consume(
             $queue,
             $this->consumerTag,

src/LaravelRabbitQueueServiceProvider.php

@@ -4,7 +4,6 @@
 
 use iamfarhad\LaravelRabbitMQ\Connectors\RabbitMQConnector;
 use iamfarhad\LaravelRabbitMQ\Console\ConsumeCommand;
-// Added missing import
 use Illuminate\Contracts\Debug\ExceptionHandler;
 use Illuminate\Support\ServiceProvider;