SWR-15367 #comment Bandwidth Splitting

Author: NekloShcherbyna

README.md

@@ -11,14 +11,14 @@ Only the latest version will get new features. Bug fixes will be provided using
 
 | Package Version | Laravel Version | Bug Fixes Until     |                                                                                             |
 |-----------------|-----------------|---------------------|---------------------------------------------------------------------------------------------|
-| 1               | 10              | December 26th, 2024 | [Documentation](https://github.com/vyuldashev/laravel-queue-rabbitmq/blob/master/README.md) |
+| 1               | 11              | December 26th, 2024 | [Documentation](https://github.com/vyuldashev/laravel-queue-rabbitmq/blob/master/README.md) |
 
 ## Installation
 
 You can install this package via composer using this command:
 
 ```
-composer require salesmessage/php-lib-rabbitmq:^1.10 --ignore-platform-reqs
+composer require salesmessage/php-lib-rabbitmq:^1.11 --ignore-platform-reqs
 ```
 
 The package will automatically register itself.

composer.json

@@ -34,7 +34,7 @@
     },
     "extra": {
         "branch-alias": {
-            "dev-master": "1.10-dev"
+            "dev-master": "1.11-dev"
         },
         "laravel": {
             "providers": [

src/VhostsConsumer.php

@@ -110,6 +110,8 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
     {
         $this->goAheadOrWait();
 
+        $this->connectionMutex = new Mutex(false);
+
         $this->configConnectionName = (string) $connectionName;
         $this->workerOptions = $options;
 
@@ -121,9 +123,7 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
 
         [$startTime, $jobsProcessed] = [hrtime(true) / 1e9, 0];
 
-        $connection = $this->initConnection();
-
-        $this->startConsuming();
+        $connection = $this->startConsuming();
 
         while ($this->channel->is_consuming()) {
             // Before reserving any jobs, we will make sure this queue is not paused and
@@ -175,9 +175,10 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
             if (false === $this->hasJob) {
                 $this->output->info('Consuming sleep. No job...');
 
+                $this->stopConsuming();
+
                 $this->processBatch($connection);
 
-                $this->stopConsuming();
                 $this->goAheadOrWait();
                 $this->startConsuming();
 
@@ -231,7 +232,11 @@ protected function getStopStatus(
         };
     }
 
-    private function startConsuming()
+    /**
+     * @return RabbitMQQueue
+     * @throws Exceptions\MutexTimeout
+     */
+    private function startConsuming(): RabbitMQQueue
     {
         $this->output->info(sprintf(
             'Start consuming. Vhost: "%s". Queue: "%s"',
@@ -276,9 +281,10 @@ private function startConsuming()
             ]);
 
             if ($jobsProcessed >= $this->batchSize) {
+                $this->stopConsuming();
+
                 $this->processBatch($connection);
 
-                $this->stopConsuming();
                 $this->goAheadOrWait();
                 $this->startConsuming();
             }
@@ -294,7 +300,7 @@ private function startConsuming()
         try {
             $this->channel->basic_consume(
                 $this->currentQueueName,
-                $this->consumerTag,
+                $this->getTagName(),
                 false,
                 false,
                 false,
@@ -328,9 +334,13 @@ private function startConsuming()
         $this->updateLastProcessedAt();
 
         if (false === $isSuccess) {
+            $this->stopConsuming();
+            
             $this->goAheadOrWait();
             return $this->startConsuming();
         }
+
+        return $connection;
     }
 
     /**
@@ -507,10 +517,10 @@ private function ackMessage(AMQPMessage $message, bool $multiple = false): void
      * @return void
      * @throws Exceptions\MutexTimeout
      */
-    private function stopConsuming()
+    private function stopConsuming(): void
     {
         $this->connectionMutex->lock(self::MAIN_HANDLER_LOCK);
-        $this->channel->basic_cancel($this->consumerTag, true);
+        $this->channel->basic_cancel($this->getTagName(), true);
         $this->connectionMutex->unlock(self::MAIN_HANDLER_LOCK);
     }
 
@@ -717,7 +727,7 @@ private function updateLastProcessedAt()
     /**
      * @return RabbitMQQueue
      */
-    private function initConnection()
+    private function initConnection(): RabbitMQQueue
     {
         $connection = $this->manager->connection(
             ConnectionNameDto::getVhostConnectionName($this->currentVhostName,  $this->configConnectionName)
@@ -751,19 +761,26 @@ private function initConnection()
         }
 
         $this->currentConnectionName = $connection->getConnectionName();
-        $this->channel = $channel;
-
-        $this->connectionMutex = new Mutex(false);
 
         $this->connectionMutex->lock(self::MAIN_HANDLER_LOCK);
-        $this->channel->basic_qos(
+        $channel->basic_qos(
             $this->prefetchSize,
             $this->prefetchCount,
             false
         );
         $this->connectionMutex->unlock(self::MAIN_HANDLER_LOCK);
 
+        $this->channel = $channel;
+
         return $connection;
     }
+
+    /**
+     * @return string
+     */
+    private function getTagName(): string
+    {
+        return $this->consumerTag . '_' .  $this->currentVhostName;
+    }
 }