Added prefetch options in consumer command

Async/AMQPAdapter.php

@@ -162,20 +162,22 @@ public function enqueue($command): PromiseInterface
      * @param CommandBus    $bus
      * @param int           $limit
      * @param OutputPrinter $outputPrinter
+     * @param Prefetch      $prefetch
      *
      * @throws InvalidCommandException
      */
     public function consume(
         CommandBus $bus,
         int $limit,
-        OutputPrinter $outputPrinter
+        OutputPrinter $outputPrinter,
+        Prefetch $prefetch
     ) {
         $this->resetIterations($limit);
         $forced = false;
 
         $this
             ->channel
-            ->qos(0, 1, true)
+            ->qos($prefetch->getPrefetchSize(), $prefetch->getPrefetchCount(), $prefetch->isGlobal())
             ->then(function () use ($bus, $outputPrinter, &$forced) {
                 return $this
                     ->channel

Async/AsyncAdapter.php

@@ -93,13 +93,15 @@ abstract public function enqueue($command): PromiseInterface;
      * @param CommandBus    $bus
      * @param int           $limit
      * @param OutputPrinter $outputPrinter
+     * @param Prefetch      $prefetch
      *
      * @throws InvalidCommandException
      */
     abstract public function consume(
         CommandBus $bus,
         int $limit,
-        OutputPrinter $outputPrinter
+        OutputPrinter $outputPrinter,
+        Prefetch $prefetch
     );
 
     /**

Async/InMemoryAdapter.php

@@ -130,13 +130,15 @@ public function enqueue($command): PromiseInterface
      * @param CommandBus    $bus
      * @param int           $limit
      * @param OutputPrinter $outputPrinter
+     * @param Prefetch      $prefetch
      *
      * @throws InvalidCommandException
      */
     public function consume(
         CommandBus $bus,
         int $limit,
-        OutputPrinter $outputPrinter
+        OutputPrinter $outputPrinter,
+        Prefetch $prefetch
     ) {
         $this->resetIterations($limit);
 

Async/PostgreSQLAdapter.php

@@ -207,13 +207,15 @@ public function enqueue($command): PromiseInterface
      * @param CommandBus    $bus
      * @param int           $limit
      * @param OutputPrinter $outputPrinter
+     * @param Prefetch      $prefetch
      *
      * @throws InvalidCommandException
      */
     public function consume(
         CommandBus $bus,
         int $limit,
-        OutputPrinter $outputPrinter
+        OutputPrinter $outputPrinter,
+        Prefetch $prefetch
     ) {
         $this->resetIterations($limit);
         $keepChecking = $this->consumeAvailableElements($bus, $outputPrinter);

Async/Prefetch.php

@@ -0,0 +1,65 @@
+<?php
+
+/*
+ * This file is part of the DriftPHP Project
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * Feel free to edit as you please, and have fun.
+ *
+ * @author Marc Morera <yuhu@mmoreram.com>
+ */
+
+declare(strict_types=1);
+
+namespace Drift\CommandBus\Async;
+
+/**
+ * Class Prefetch.
+ */
+class Prefetch
+{
+    private int $prefetchSize;
+    private int $prefetchCount;
+    private bool $global;
+
+    /**
+     * @param int  $prefetchSize
+     * @param int  $prefetchCount
+     * @param bool $global
+     */
+    public function __construct(
+        int $prefetchSize,
+        int $prefetchCount,
+        bool $global
+    ) {
+        $this->prefetchSize = $prefetchSize;
+        $this->prefetchCount = $prefetchCount;
+        $this->global = $global;
+    }
+
+    /**
+     * @return int
+     */
+    public function getPrefetchSize(): int
+    {
+        return $this->prefetchSize;
+    }
+
+    /**
+     * @return int
+     */
+    public function getPrefetchCount(): int
+    {
+        return $this->prefetchCount;
+    }
+
+    /**
+     * @return bool
+     */
+    public function isGlobal(): bool
+    {
+        return $this->global;
+    }
+}

Async/RedisAdapter.php

@@ -134,13 +134,15 @@ public function enqueue($command): PromiseInterface
      * @param CommandBus    $bus
      * @param int           $limit
      * @param OutputPrinter $outputPrinter
+     * @param Prefetch      $prefetch
      *
      * @throws InvalidCommandException
      */
     public function consume(
         CommandBus $bus,
         int $limit,
-        OutputPrinter $outputPrinter
+        OutputPrinter $outputPrinter,
+        Prefetch $prefetch
     ) {
         $this->resetIterations($limit);
 

Console/CommandConsumerCommand.php

@@ -16,6 +16,7 @@
 namespace Drift\CommandBus\Console;
 
 use Drift\CommandBus\Async\AsyncAdapter;
+use Drift\CommandBus\Async\Prefetch;
 use Drift\CommandBus\Bus\InlineCommandBus;
 use Drift\Console\OutputPrinter;
 use Drift\EventBus\Bus\EventBus;
@@ -78,6 +79,10 @@ protected function configure()
                 'Exchanges to listen'
             );
         }
+
+        $this->addOption('prefetch-size', null, InputOption::VALUE_OPTIONAL, 'Prefetch size. Only for AMQP', 0);
+        $this->addOption('prefetch-count', null, InputOption::VALUE_OPTIONAL, 'Prefetch count. Only for AMQP', 1);
+        $this->addOption('is-prefetch-local', null, InputOption::VALUE_NONE, 'Prefetch is global. Only for AMQP');
     }
 
     /**
@@ -113,12 +118,19 @@ class_exists(EventBusSubscriber::class) &&
                 );
         }
 
+        $prefetch = new Prefetch(
+            \intval($input->getOption('prefetch-size')),
+            \intval($input->getOption('prefetch-count')),
+            !\boolval($input->getOption('is-prefetch-local')),
+        );
+
         $this
             ->asyncAdapter
             ->consume(
                 $this->commandBus,
                 \intval($input->getOption('limit')),
-                $outputPrinter
+                $outputPrinter,
+                $prefetch
             );
 
         return 0;