SWR-15367

Author: NekloShcherbyna

src/Console/ConsumeVhostsCommand.php

@@ -17,6 +17,7 @@ class ConsumeVhostsCommand extends WorkCommand
                             {--name=default : The name of the consumer}
                             {--vhosts= : The list of the vhosts to work}
                             {--queues= : The list of the queues to work}
+                            {--batch-size=100 : The number of jobs for batch}
                             {--once : Only process the next job on the queue}
                             {--stop-when-empty : Stop when the queue is empty}
                             {--delay=0 : The number of seconds to delay failed jobs (Deprecated)}
@@ -51,6 +52,7 @@ 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->setBatchSize((int) $this->option('batch-size'));
 
         $filtersDto = new ConsumeVhostsFiltersDto(
             trim($this->option('vhosts', '')),

src/VhostsConsumer.php

@@ -27,12 +27,12 @@ class VhostsConsumer extends Consumer
 {
     protected const MAIN_HANDLER_LOCK = 'vhost_handler';
 
-    protected const CONSUME_BATCH_SIZE = 5;
-
     private ?OutputStyle $output = null;
 
     private ?ConsumeVhostsFiltersDto $filtersDto = null;
 
+    private int $batchSize = 100;
+
     private string $configConnectionName = '';
 
     private string $currentConnectionName = '';
@@ -91,15 +91,19 @@ public function setFiltersDto(ConsumeVhostsFiltersDto $filtersDto): self
         return $this;
     }
 
-    public function daemon($connectionName, $queue, WorkerOptions $options)
+    /**
+     * @param int $batchSize
+     * @return $this
+     */
+    public function setBatchSize(int $batchSize): self
     {
-        $this->loadVhosts();
-        if (false === $this->switchToNextVhost()) {
-            // @todo load vhosts again
-            $this->output->warning('No active vhosts... Exit');
+        $this->batchSize = $batchSize;
+        return $this;
+    }
 
-            return;
-        }
+    public function daemon($connectionName, $queue, WorkerOptions $options)
+    {
+        $this->goAheadOrWait();
 
         $this->configConnectionName = (string) $connectionName;
         $this->workerOptions = $options;
@@ -259,7 +263,7 @@ private function startConsuming()
                 $jobsProcessed
             ));
 
-            if ($jobsProcessed >= self::CONSUME_BATCH_SIZE) {
+            if ($jobsProcessed >= $this->batchSize) {
                 $this->processBatch($connection);
 
                 $this->stopConsuming();