Merge pull request #17 from ensi-platform/devops-2512

DEVOPS-2512 feature: listen multiple topics

Author: MsNatali

README.md

@@ -166,7 +166,7 @@ use Ensi\LaravelPhpRdKafkaConsumer\Commands\KafkaConsumeCommand;
 use Ensi\LaravelPhpRdKafkaConsumer\Tests\ConsumerFaker;
 use RdKafka\Message;
 
-ConsumerFaker::new('test-model')
+ConsumerFaker::new(['test-model'])
     ->addMessage(new Message())
     ->addMessage(new Message())
     ->consume();

phpunit.xml

@@ -1,20 +1,16 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:noNamespaceSchemaLocation="./vendor/phpunit/phpunit/phpunit.xsd"
-         bootstrap="vendor/autoload.php"
-         colors="true"
->
-    <testsuites>
-        <testsuite name="Tests">
-            <directory>tests</directory>
-        </testsuite>
-    </testsuites>
-    <coverage>
-        <include>
-            <directory suffix="Test.php">./src</directory>
-        </include>
-    </coverage>
-    <logging>
-        <junit outputFile="build/report.junit.xml"/>
-    </logging>
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/10.5/phpunit.xsd" bootstrap="vendor/autoload.php" colors="true">
+  <testsuites>
+    <testsuite name="Tests">
+      <directory>tests</directory>
+    </testsuite>
+  </testsuites>
+  <logging>
+    <junit outputFile="build/report.junit.xml"/>
+  </logging>
+  <source>
+    <include>
+      <directory suffix="Test.php">./src</directory>
+    </include>
+  </source>
 </phpunit>

src/Commands/KafkaConsumeCommand.php

@@ -18,7 +18,7 @@ class KafkaConsumeCommand extends Command implements SignalableCommandInterface
      * The name and signature of the console command.
      */
     protected $signature = 'kafka:consume
-                            {topic-key : The key of a topic in the kafka.topics list}
+                            {topic-key : The key of a topic in the kafka.topics list, can accept multiple keys delimited by comma}
                             {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}
@@ -82,14 +82,18 @@ public function handleSignal(int $signal, int|false $previousExitCode = 0): int|
      */
     public function handle(ConsumerFactory $consumerFactory): int
     {
+        $topicKeys = explode(',', $this->getTopicKey());
+
         try {
             $this->consumer = $consumerFactory
-                ->build($this->getTopicKey(), $this->getConsumerName())
+                ->build($topicKeys, $this->getConsumerName())
                 ->setMaxEvents($this->getMaxEvents())
                 ->setMaxTime($this->getMaxTime());
 
-            $this->info("Start listening to topic: \"{$this->getTopicKey()}\"" .
-                " ({$this->consumer->getTopicName()}), consumer \"{$this->getConsumerName()}\"");
+            foreach ($this->consumer->getTopicNames() as $topicName) {
+                $this->info("Start listening to topic:" .
+                    " {$topicName}, consumer \"{$this->getConsumerName()}\"");
+            }
 
             $this->consumer->listen();
         } catch (Throwable $exception) {

src/Consumers/Consumer.php

@@ -5,22 +5,28 @@
 use Ensi\LaravelPhpRdKafkaConsumer\ConsumerOptions;
 use Ensi\LaravelPhpRdKafkaConsumer\HighLevelConsumer;
 use Ensi\LaravelPhpRdKafkaConsumer\ProcessorData;
-use RdKafka\Exception;
 use Throwable;
 
 class Consumer
 {
+    /**
+     * @param HighLevelConsumer $highLevelConsumer
+     * @param ProcessorData[] $processorData
+     * @param ConsumerOptions $consumerOptions
+     * @param array $topicNames
+     */
     public function __construct(
         protected HighLevelConsumer $highLevelConsumer,
-        protected ProcessorData $processorData,
+        protected array $processorData,
         protected ConsumerOptions $consumerOptions,
-        protected string $topicName
+        protected array $topicNames,
+        protected string $consumerName,
     ) {
     }
 
-    public function getTopicName(): string
+    public function getTopicNames(): array
     {
-        return $this->topicName;
+        return $this->topicNames;
     }
 
     public function setMaxTime(int $maxTime = 0): self
@@ -42,24 +48,18 @@ public function forceStop(): void
         $this->highLevelConsumer->forceStop();
     }
 
-    public function getProcessorData(): ProcessorData
-    {
-        return $this->processorData;
-    }
-
     public function getConsumerOptions(): ConsumerOptions
     {
         return $this->consumerOptions;
     }
 
     /**
-     * @throws Exception
      * @throws Throwable
      */
     public function listen(): void
     {
         $this->highLevelConsumer
-            ->for($this->processorData->consumer)
-            ->listen($this->topicName, $this->processorData, $this->consumerOptions);
+            ->for($this->consumerName)
+            ->listen($this->topicNames, $this->processorData, $this->consumerOptions);
     }
 }

src/Consumers/Factories/ConsumerFactory.php

@@ -19,16 +19,24 @@ public function __construct(
     /**
      * @throws KafkaConsumerProcessorException
      */
-    public function build(string $topicKey, string $consumerName = 'default'): Consumer
+    public function build(array $topicKeys, string $consumerName = 'default'): Consumer
     {
-        $processorData = $this->makeProcessorData($topicKey, $consumerName);
-        $consumerOptions = $this->makeConsumerOptions($consumerName, $processorData);
+        $topicNames = [];
+        $processors = [];
+        foreach ($topicKeys as $topicKey) {
+            $topicName = KafkaFacade::topicNameByClient('consumer', $consumerName, $topicKey);
+            $topicNames[] = $topicName;
+            $processors[$topicName] = $this->makeProcessorData($topicKey, $consumerName);
+        }
+
+        $consumerOptions = $this->makeConsumerOptions($consumerName);
 
         return new Consumer(
             highLevelConsumer: $this->highLevelConsumer,
-            processorData: $processorData,
+            processorData: $processors,
             consumerOptions: $consumerOptions,
-            topicName: KafkaFacade::topicNameByClient('consumer', $consumerName, $topicKey)
+            topicNames: $topicNames,
+            consumerName: $consumerName,
         );
     }
 
@@ -39,10 +47,6 @@ protected function makeProcessorData(string $topicKey, string $consumerName): Pr
     {
         $processorData = $this->findMatchedProcessor($topicKey, $consumerName);
 
-        if (!class_exists($processorData->class)) {
-            throw new KafkaConsumerProcessorException("Processor class \"$processorData->class\" is not found");
-        }
-
         if (!$processorData->hasValidType()) {
             throw new KafkaConsumerProcessorException("Invalid processor type \"$processorData->type\"," .
                 " supported types are: " . implode(',', $processorData->getSupportedTypes()));
@@ -75,12 +79,12 @@ class: $processor['class'],
         throw new KafkaConsumerProcessorException("Processor for topic-key \"$topicKey\" and consumer \"$consumerName\" is not found");
     }
 
-    protected function makeConsumerOptions(string $consumerName, ProcessorData $processorData): ConsumerOptions
+    protected function makeConsumerOptions(string $consumerName): ConsumerOptions
     {
         $consumerPackageOptions = config('kafka-consumer.consumer_options.' . $consumerName, []);
 
         return new ConsumerOptions(
-            consumeTimeout: $consumerPackageOptions['consume_timeout'] ?? $processorData->consumeTimeout,
+            consumeTimeout: $consumerPackageOptions['consume_timeout'] ?? 20000,
             middleware: $this->collectMiddleware($consumerPackageOptions['middleware'] ?? []),
         );
     }

src/HighLevelConsumer.php

@@ -11,7 +11,6 @@
 use RdKafka\KafkaConsumer;
 use RdKafka\Message;
 use RdKafka\TopicPartition;
-use Throwable;
 
 class HighLevelConsumer
 {
@@ -42,14 +41,14 @@ public function forceStop(): static
     }
 
     /**
-     * @throws RdKafkaException
-     * @throws Throwable
-     *
-     * @noinspection PhpRedundantCatchClauseInspection
+     * @param array $topicNames
+     * @param ProcessorData[] $processors
+     * @param ConsumerOptions $options
+     * @throws KafkaConsumerException
      */
-    public function listen(string $topicName, ProcessorData $processorData, ConsumerOptions $options): void
+    public function listen(array $topicNames, array $processors, ConsumerOptions $options): void
     {
-        $this->subscribe($topicName);
+        $this->subscribe($topicNames);
 
         [$startTime, $eventsProcessed] = [hrtime(true) / 1e9, 0];
 
@@ -60,7 +59,11 @@ public function listen(string $topicName, ProcessorData $processorData, Consumer
                 switch ($message->err) {
 
                     case RD_KAFKA_RESP_ERR_NO_ERROR:
-                        $this->processThroughMiddleware($processorData, $message, $options);
+                        if (!isset($processors[$message->topic_name])) {
+                            throw new KafkaConsumerException("Kafka error: no processor defined for topic '{$message->topic_name}'");
+                        }
+
+                        $this->processThroughMiddleware($processors[$message->topic_name], $message, $options);
                         $this->consumer->commitAsync($message);
                         $eventsProcessed++;
 
@@ -137,19 +140,21 @@ protected function shouldBeStopped(int|float $startTime, int $eventsProcessed, C
         return $this->forceStop;
     }
 
-    protected function subscribe(string $topicName): void
+    protected function subscribe(array $topicNames): void
     {
-        $attempts = 0;
-        do {
-            $topicExists = (bool)$this->getPartitions($topicName);
-            if (!$topicExists) {
-                $this->consumer->newTopic($topicName);
-                sleep(1);
-            }
-            $attempts++;
-        } while (!$topicExists && $attempts < 10);
+        foreach ($topicNames as $topicName) {
+            $attempts = 0;
+            do {
+                $topicExists = (bool)$this->getPartitions($topicName);
+                if (!$topicExists) {
+                    $this->consumer->newTopic($topicName);
+                    sleep(1);
+                }
+                $attempts++;
+            } while (!$topicExists && $attempts < 10);
+        }
 
-        $this->consumer->subscribe([$topicName]);
+        $this->consumer->subscribe($topicNames);
     }
 
     /**

src/Tests/Consumer/KafkaConsumer.php

@@ -12,11 +12,11 @@ class KafkaConsumer extends BaseKafkaConsumer
 {
     protected Metadata $metadata;
 
-    public function __construct(string $topicName, protected array $messages = [])
+    public function __construct(array $topicNames, protected array $messages = [])
     {
         parent::__construct($this->makeConf());
 
-        $this->metadata = new Metadata($topicName);
+        $this->metadata = new Metadata($topicNames);
     }
 
     private function makeConf(): Conf
@@ -42,7 +42,7 @@ public function consume($timeout_ms)
     /**
      * @phpstan-ignore-next-line
      */
-    public function getMetadata($all_topics, $only_topic = null, $timeout_ms): Metadata
+    public function getMetadata($all_topics, $only_topic = null, $timeout_ms = 0): Metadata
     {
         return $this->metadata;
     }

src/Tests/Consumer/Topics/Metadata.php

@@ -2,13 +2,15 @@
 
 namespace Ensi\LaravelPhpRdKafkaConsumer\Tests\Consumer\Topics;
 
+use Illuminate\Support\Arr;
+
 class Metadata
 {
     protected array $topics = [];
 
-    public function __construct(string $topicName)
+    public function __construct(array $topicNames)
     {
-        $this->topics[] = new Topic($topicName);
+        $this->topics = Arr::map($topicNames, fn ($topicName) => new Topic($topicName));
     }
 
     public function getTopics(): array

src/Tests/ConsumerFaker.php

@@ -9,20 +9,21 @@
 use Ensi\LaravelPhpRdKafkaConsumer\HighLevelConsumer;
 use Ensi\LaravelPhpRdKafkaConsumer\Tests\Consumer\KafkaConsumer;
 use Ensi\LaravelPhpRdKafkaConsumer\Tests\Exceptions\OnlyTestingEnvironmentException;
+use Illuminate\Support\Arr;
 use RdKafka\Message;
 use Throwable;
 
 class ConsumerFaker
 {
     protected array $messages = [];
 
-    protected string $topicName;
+    protected array $topicNames;
 
     public function __construct(
-        protected string $topicKey,
+        protected array  $topicKeys,
         protected string $consumerName = 'default'
     ) {
-        $this->topicName = KafkaFacade::topicNameByClient('consumer', $consumerName, $topicKey);
+        $this->topicNames = Arr::map($topicKeys, fn ($topicKey) => KafkaFacade::topicNameByClient('consumer', $consumerName, $topicKey));
     }
 
     public function addMessage(Message $message): self
@@ -48,7 +49,7 @@ public function consume(): void
         $this->bind();
 
         (new ConsumerFactory(resolve(HighLevelConsumer::class)))
-            ->build($this->topicKey, $this->consumerName)
+            ->build($this->topicKeys, $this->consumerName)
             ->listen();
     }
 
@@ -66,11 +67,11 @@ public function bind(): void
 
     private function makeKafkaManager(): KafkaManager
     {
-        return new KafkaManager(new KafkaConsumer($this->topicName, $this->messages));
+        return new KafkaManager(new KafkaConsumer($this->topicNames, $this->messages));
     }
 
-    public static function new(string $topicKey, string $consumerName = 'default'): self
+    public static function new(array $topicKeys, string $consumerName = 'default'): self
     {
-        return new self($topicKey, $consumerName);
+        return new self($topicKeys, $consumerName);
     }
 }