SWR-15367

Author: NekloShcherbyna

README.md

@@ -9,16 +9,16 @@ RabbitMQ Queue driver for Laravel
 
 Only the latest version will get new features. Bug fixes will be provided using the following scheme:
 
-| Package Version | Laravel Version | Bug Fixes Until   |                                                                                             |
-|-----------------|-----------------|-------------------|---------------------------------------------------------------------------------------------|
-| 1               | 06              | December 24th, 2024 | [Documentation](https://github.com/vyuldashev/laravel-queue-rabbitmq/blob/master/README.md) |
+| Package Version | Laravel Version | Bug Fixes Until     |                                                                                             |
+|-----------------|-----------------|---------------------|---------------------------------------------------------------------------------------------|
+| 1               | 07              | 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.06 --ignore-platform-reqs
+composer require salesmessage/php-lib-rabbitmq:^1.07 --ignore-platform-reqs
 ```
 
 The package will automatically register itself.
@@ -462,6 +462,26 @@ It is possible to change te default queue by adding an extra parameter in the co
 ],
 ```
 
+### Immediate Indexation
+
+By default, your connection will be created with a immediate indexation setting of `false`.
+
+
+```php
+
+'connections' => [
+    // ...
+
+    'rabbitmq_vhosts' => [
+        // ...
+
+        'immediate_indexation' => env('RABBITMQ_IMMEDIATE_INDEXATION', false),
+    ],
+
+    // ...    
+],
+```
+
 ### Heartbeat
 
 By default, your connection will be created with a heartbeat setting of `0`.
@@ -574,7 +594,7 @@ Available protocols : `tcp`, `ssl`, `tls`
 
 ### Octane support
 
-Starting with 13.3.0, this package supports [Laravel Octane](https://laravel.com/docs/octane) out of the box.
+Starting with 1.04, this package supports [Laravel Octane](https://laravel.com/docs/octane) out of the box.
 Firstly, install Octane and don't forget to warm 'rabbitmq' connection in the octane config.
 > See: https://github.com/vyuldashev/laravel-queue-rabbitmq/issues/460#issuecomment-1469851667
 

composer.json

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

config/rabbitmq.php

@@ -10,6 +10,7 @@
     'driver' => 'rabbitmq_vhosts',
     'queue' => env('RABBITMQ_QUEUE', 'default'),
     'connection' => 'default',
+    'immediate_indexation' => env('RABBITMQ_IMMEDIATE_INDEXATION', false),
 
     'hosts' => [
         [

src/Dto/QueueApiDto.php

@@ -55,6 +55,16 @@ public function getMessages(): int
         return $this->messages;
     }
 
+    /**
+     * @param int $messages
+     * @return $this
+     */
+    public function setMessages(int $messages): self
+    {
+        $this->messages = $messages;
+        return $this;
+    }
+
     /**
      * @return int
      */
@@ -63,6 +73,16 @@ public function getMessagesReady(): int
         return $this->messagesReady;
     }
 
+    /**
+     * @param int $messagesReady
+     * @return $this
+     */
+    public function setMessagesReady(int $messagesReady): self
+    {
+        $this->messagesReady = $messagesReady;
+        return $this;
+    }
+
     /**
      * @return int
      */

src/Dto/VhostApiDto.php

@@ -52,6 +52,16 @@ public function getMessages(): int
         return $this->messages;
     }
 
+    /**
+     * @param int $messages
+     * @return $this
+     */
+    public function setMessages(int $messages): self
+    {
+        $this->messages = $messages;
+        return $this;
+    }
+
     /**
      * @return int
      */
@@ -60,6 +70,16 @@ public function getMessagesReady(): int
         return $this->messagesReady;
     }
 
+    /**
+     * @param int $messagesReady
+     * @return $this
+     */
+    public function setMessagesReady(int $messagesReady): self
+    {
+        $this->messagesReady = $messagesReady;
+        return $this;
+    }
+
     /**
      * @return int
      */

src/Queue/RabbitMQQueueBatchable.php

@@ -4,15 +4,44 @@
 
 use PhpAmqpLib\Connection\AbstractConnection;
 use Salesmessage\LibRabbitMQ\Dto\ConnectionNameDto;
+use Salesmessage\LibRabbitMQ\Dto\QueueApiDto;
+use Salesmessage\LibRabbitMQ\Dto\VhostApiDto;
 use Salesmessage\LibRabbitMQ\Queue\RabbitMQQueue as BaseRabbitMQQueue;
 use PhpAmqpLib\Exception\AMQPChannelClosedException;
 use PhpAmqpLib\Exception\AMQPConnectionClosedException;
 use PhpAmqpLib\Channel\AMQPChannel;
+use Salesmessage\LibRabbitMQ\Services\GroupsService;
+use Salesmessage\LibRabbitMQ\Services\InternalStorageManager;
 use Salesmessage\LibRabbitMQ\Services\VhostsService;
 
 class RabbitMQQueueBatchable extends BaseRabbitMQQueue
 {
-    protected function publishBasic($msg, $exchange = '', $destination = '', $mandatory = false, $immediate = false, $ticket = null): void
+    private InternalStorageManager $internalStorageManager;
+
+    private GroupsService $groupsService;
+
+    private VhostsService $vhostsService;
+
+    /**
+     * @param QueueConfig $config
+     */
+    public function __construct(QueueConfig $config)
+    {
+        $this->internalStorageManager = app(InternalStorageManager::class);
+        $this->groupsService = app(GroupsService::class);
+        $this->vhostsService = app(VhostsService::class);
+
+        parent::__construct($config);
+    }
+
+    protected function publishBasic(
+        $msg,
+        $exchange = '',
+        $destination = '',
+        $mandatory = false,
+        $immediate = false,
+        $ticket = null
+    ): void
     {
         try {
             parent::publishBasic($msg, $exchange, $destination, $mandatory, $immediate, $ticket);
@@ -61,6 +90,10 @@ public function push($job, $data = '', $queue = null)
             $result = parent::push($job, $data, $queue);
         }
 
+        if (config('queue.connections.rabbitmq_vhosts.immediate_indexation')) {
+            $this->addQueueToIndex((string) $queue);
+        }
+
         return $result;
     }
 
@@ -81,10 +114,39 @@ private function createNotExistsVhost(): bool
             return false;
         }
 
-        /** @var VhostsService $vhostsService */
-        $vhostsService = app(VhostsService::class);
-
-        return $vhostsService->createVhost($dto->getVhostName(), 'Automatically created vhost');
+        return $this->vhostsService->createVhost($dto->getVhostName(), 'Automatically created vhost');
     }
 
+    /**
+     * @param string $queue
+     * @return bool
+     */
+    private function addQueueToIndex(string $queue): bool
+    {
+        if ('' === $queue) {
+            return false;
+        }
+
+        $dto = new ConnectionNameDto($this->getConnectionName());
+        $vhostName = $dto->getVhostName();
+        if (null === $vhostName) {
+            return false;
+        }
+
+        $groups = $this->groupsService->getAllGroupsNames();
+
+        $queueApiDto = new QueueApiDto([
+            'name' => $queue,
+            'vhost' => $vhostName,
+        ]);
+        $isQueueActivated = $this->internalStorageManager->activateQueue($queueApiDto, $groups);
+
+        $vhostDto = new VhostApiDto([
+            'name' => $vhostName,
+        ]);
+        $isVhostActivated = $this->internalStorageManager->activateVhost($vhostDto, $groups);
+
+        return $isQueueActivated && $isVhostActivated;
+    }
 }
+

src/Services/InternalStorageManager.php

@@ -121,6 +121,24 @@ public function removeVhost(VhostApiDto $vhostDto): bool
         return true;
     }
 
+    /**
+     * @param VhostApiDto $vhostDto
+     * @param array $groups
+     * @return bool
+     */
+    public function activateVhost(VhostApiDto $vhostDto, array $groups): bool
+    {
+        $storageKey = $this->getVhostStorageKey($vhostDto);
+
+        $messages = (int) $this->redis->hget($storageKey, 'messages') + 1;
+        $vhostDto->setMessages($messages);
+
+        $messagesReady = (int) $this->redis->hget($storageKey, 'messages_ready') + 1;
+        $vhostDto->setMessagesReady($messagesReady);
+
+        return $this->addVhost($vhostDto, $groups);
+    }
+
     /**
      * @param VhostApiDto $vhostDto
      * @return bool
@@ -215,6 +233,24 @@ public function removeQueue(QueueApiDto $queueDto): bool
         return true;
     }
 
+    /**
+     * @param QueueApiDto $queueDto
+     * @param array $groups
+     * @return bool
+     */
+    public function activateQueue(QueueApiDto $queueDto, array $groups): bool
+    {
+        $storageKey = $this->getQueueStorageKey($queueDto);
+
+        $messages = (int) $this->redis->hget($storageKey, 'messages') + 1;
+        $queueDto->setMessages($messages);
+
+        $messagesReady = (int) $this->redis->hget($storageKey, 'messages_ready') + 1;
+        $queueDto->setMessagesReady($messagesReady);
+
+        return $this->addQueue($queueDto, $groups);
+    }
+
     /**
      * @param VhostApiDto $vhostDto
      * @return bool