Merge pull request #2 from ensi-platform/task-82065

#82065 more confi options of consumer

Author: arrilot

README.md

@@ -20,7 +20,7 @@ Now go to `config/kafka-consumer.php` and add processors there.
 
 ## Usage
 
-The package provides `php artisan kafka:consume {topic} {consumer=default}` command that executes the first processor that matches given topic and consumer name. Consumer name is taken from ensi/laravel-phprdkafka config file.
+The package provides `php artisan kafka:consume {topic} {consumer=default} {--max-events=0} {--max-time=0} {--once}` command that executes the first processor that matches given topic and consumer name. Consumer name is taken from `ensi/laravel-phprdkafka config` file.
 
 Processors in config have the following configuration options:
 

src/Commands/KafkaConsumeCommand.php

@@ -2,6 +2,7 @@
 
 namespace Ensi\LaravelPhpRdKafkaConsumer\Commands;
 
+use Ensi\LaravelPhpRdKafkaConsumer\ConsumerOptions;
 use Ensi\LaravelPhpRdKafkaConsumer\HighLevelConsumer;
 use Throwable;
 use Illuminate\Console\Command;
@@ -11,7 +12,13 @@ class KafkaConsumeCommand extends Command
     /**
      * The name and signature of the console command.
      */
-    protected $signature = 'kafka:consume {topic} {consumer=default}';
+    protected $signature = 'kafka:consume
+                            {topic : The name of the topic}
+                            {consumer=default : The name of the consumer}
+                            {--max-events=0 : The number of events to consume before stopping}
+                            {--max-time=0 : The maximum number of seconds the worker should run}
+                            {--once : Only process the next event in the topic}
+                            ';
 
     /**
      * The console command description.
@@ -50,8 +57,6 @@ public function handle(): int
             return 1;
         }
 
-        $consumeTimeout = $processorData['consume_timeout'] ?? 20000;
-
         $supportedProcessorTypes = ['action', 'job'];
         $processorType = $processorData['type'] ?? 'action';
         if (!in_array($processorType, $supportedProcessorTypes)) {
@@ -62,9 +67,15 @@ public function handle(): int
 
         $processorQueue = $processorData['queue'] ?? false;
 
+        $consumerOptions = new ConsumerOptions(
+            consumeTimeout: $processorData['consume_timeout'] ?? 20000,
+            maxEvents: $this->option('once') ? 1 : (int) $this->option('max-events'),
+            maxTime: (int) $this->option('max-time')
+        );
+
         $this->info("Start listenning to topic: \"$topic\", consumer \"$consumer\"");
         try {
-            $kafkaTopicListener = new HighLevelConsumer($topic, $consumer, $consumeTimeout);
+            $kafkaTopicListener = new HighLevelConsumer($topic, $consumer, $consumerOptions);
             $kafkaTopicListener->listen($processorClassName, $processorType, $processorQueue);
         } catch (Throwable $e) {
             $this->error('An error occurred while listening to the topic: '. $e->getMessage(). ' '. $e->getFile() . '::' . $e->getLine());

src/ConsumerOptions.php

@@ -0,0 +1,14 @@
+<?php
+
+namespace Ensi\LaravelPhpRdKafkaConsumer;
+
+class ConsumerOptions
+{
+    public function __construct(
+        public int $consumeTimeout = 20000,
+        public int $maxEvents = 0,
+        public int $maxTime = 0,
+    )
+    {
+    }
+}

src/HighLevelConsumer.php

@@ -16,7 +16,7 @@ class HighLevelConsumer
     public function __construct(
         protected string $topicName, 
         ?string $consumerName = null, 
-        protected int $consumeTimeout = 20000,
+        protected ConsumerOptions $options,
     )
     {
         $manager = resolve(KafkaManager::class);
@@ -32,14 +32,17 @@ public function listen(string $processorClassName, string $processorType, string
     {
         $this->consumer->subscribe([ $this->topicName ]);
 
+        [$startTime, $eventsProcessed] = [hrtime(true) / 1e9, 0];
+
         while (true) {
-            $message = $this->consumer->consume($this->consumeTimeout);
+            $message = $this->consumer->consume($this->options->consumeTimeout);
 
             switch ($message->err) {
 
                 case RD_KAFKA_RESP_ERR_NO_ERROR:
                     $this->executeProcessor($processorClassName, $processorType, $processorQueue, $message);
                     $this->consumer->commitAsync($message);
+                    $eventsProcessed++;
                     break;
 
                 case RD_KAFKA_RESP_ERR__TIMED_OUT:
@@ -51,6 +54,10 @@ public function listen(string $processorClassName, string $processorType, string
                 default:
                     throw new KafkaConsumerException('Kafka error: ' . $message->errstr());
             }
+
+            if ($this->shouldBeStopped($startTime, $eventsProcessed)) {
+                break;
+            }
         }
     }
 
@@ -79,4 +86,17 @@ protected function executeQueueableProcessor(string $className, string $type, st
             is_string($queue) ? $processor->onQueue($queue)->execute($message) : $processor->execute($message);
         }
     }
-}
+
+    protected function shouldBeStopped(int|float $startTime, int $eventsProcessed): bool
+    {
+        if ($this->options->maxTime && hrtime(true) / 1e9 - $startTime >= $this->options->maxTime) {
+            return true;
+        } 
+
+        if ($this->options->maxEvents && $eventsProcessed >= $this->options->maxEvents) {
+            return true;
+        }
+
+        return false;
+    }
+}