Merge pull request #8 from ensi-platform/task-102492

#102492 работа с топиками из списка

Author: MadridianFox

.gitignore

@@ -10,3 +10,4 @@ psalm.xml
 testbench.yaml
 vendor
 node_modules
+studio.json

composer.json

@@ -18,7 +18,7 @@
     "require": {
         "php": "^8.0",
         "ext-rdkafka": "*",
-        "ensi/laravel-phprdkafka": "^0.2",
+        "ensi/laravel-phprdkafka": "^0.3",
         "illuminate/contracts": "^8.37 || ^9.0",
         "illuminate/pipeline": "^8.37 || ^9.0",
         "illuminate/support": "^8.37 || ^9.0"

src/Commands/KafkaConsumeCommand.php

@@ -2,6 +2,7 @@
 
 namespace Ensi\LaravelPhpRdKafkaConsumer\Commands;
 
+use Ensi\LaravelPhpRdKafka\KafkaFacade;
 use Ensi\LaravelPhpRdKafkaConsumer\ConsumerOptions;
 use Ensi\LaravelPhpRdKafkaConsumer\HighLevelConsumer;
 use Ensi\LaravelPhpRdKafkaConsumer\ProcessorData;
@@ -15,7 +16,7 @@ class KafkaConsumeCommand extends Command implements SignalableCommandInterface
      * The name and signature of the console command.
      */
     protected $signature = 'kafka:consume
-                            {topic : The name of the topic}
+                            {topic-key : The key of a topic in the kafka.topics list}
                             {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}
@@ -53,20 +54,12 @@ public function handleSignal(int $signal): void
     public function handle(HighLevelConsumer $highLevelConsumer): int
     {
         $this->consumer = $highLevelConsumer;
-        $topic = $this->argument('topic');
+        $topicKey = $this->argument('topic-key');
         $consumer = $this->argument('consumer');
-        $availableConsumers = array_keys(config('kafka.consumers', []));
 
-        if (!in_array($consumer, $availableConsumers)) {
-            $this->error("Unknown consumer \"$consumer\"");
-            $this->line('Available consumers are: "' . implode(', ', $availableConsumers) . '" and can be found in /config/kafka.php');
-
-            return 1;
-        }
-
-        $processorData = $this->findMatchedProcessor($topic, $consumer);
+        $processorData = $this->findMatchedProcessor($topicKey, $consumer);
         if (is_null($processorData)) {
-            $this->error("Processor for topic \"$topic\" and consumer \"$consumer\" is not found");
+            $this->error("Processor for topic-key \"$topicKey\" and consumer \"$consumer\" is not found");
             $this->line('Processors are set in /config/kafka-consumers.php');
 
             return 1;
@@ -93,12 +86,13 @@ public function handle(HighLevelConsumer $highLevelConsumer): int
             middleware: $this->collectMiddleware($consumerPackageOptions['middleware'] ?? []),
         );
 
-        $this->info("Start listenning to topic: \"$topic\", consumer \"$consumer\"");
+        $topicName = KafkaFacade::topicNameByClient('consumer', $consumer, $topicKey);
+        $this->info("Start listening to topic: \"{$topicKey}\" ({$topicName}), consumer \"{$consumer}\"");
 
         try {
             $highLevelConsumer
                 ->for($consumer)
-                ->listen($topic, $processorData, $consumerOptions);
+                ->listen($topicName, $processorData, $consumerOptions);
         } catch (Throwable $e) {
             $this->error('An error occurred while listening to the topic: '. $e->getMessage(). ' '. $e->getFile() . '::' . $e->getLine());
 
@@ -111,13 +105,12 @@ public function handle(HighLevelConsumer $highLevelConsumer): int
     protected function findMatchedProcessor(string $topic, string $consumer): ?ProcessorData
     {
         foreach (config('kafka-consumer.processors', []) as $processor) {
-            if (
-                (empty($processor['topic']) || $processor['topic'] === $topic)
-                && (empty($processor['consumer']) || $processor['consumer'] === $consumer)
-                ) {
+            $topicMatched = empty($processor['topic']) || $processor['topic'] === $topic;
+            $consumerMatched = empty($processor['consumer']) || $processor['consumer'] === $consumer;
+            if ($topicMatched && $consumerMatched) {
                 return new ProcessorData(
                     class: $processor['class'],
-                    topic: $processor['topic'] ?? null,
+                    topicKey: $processor['topic'] ?? null,
                     consumer: $processor['consumer'] ?? null,
                     type: $processor['type'] ?? 'action',
                     queue: $processor['queue'] ?? false,

src/HighLevelConsumer.php

@@ -4,6 +4,7 @@
 
 use Ensi\LaravelPhpRdKafka\KafkaManager;
 use Ensi\LaravelPhpRdKafkaConsumer\Exceptions\KafkaConsumerException;
+use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Pipeline\Pipeline;
 use RdKafka\Exception as RdKafkaException;
 use RdKafka\KafkaConsumer;
@@ -15,7 +16,7 @@ class HighLevelConsumer
 {
     protected ?KafkaConsumer $consumer;
 
-    protected $forceStop = false;
+    protected bool $forceStop = false;
 
     public function __construct(
         protected KafkaManager $kafkaManager,
@@ -40,7 +41,6 @@ public function forceStop(): static
     }
 
     /**
-     * @throws KafkaException
      * @throws RdKafkaException
      * @throws Throwable
      */
@@ -95,6 +95,7 @@ protected function executeProcessor(ProcessorData $processorData, Message $messa
 
     protected function executeSyncProcessor(ProcessorData $processorData, Message $message): void
     {
+        /** @var class-string|Dispatchable $className */
         $className = $processorData->class;
         if ($processorData->type === 'job') {
             $className::dispatchSync($message);
@@ -105,6 +106,7 @@ protected function executeSyncProcessor(ProcessorData $processorData, Message $m
 
     protected function executeQueueableProcessor(ProcessorData $processorData, Message $message): void
     {
+        /** @var class-string|Dispatchable $className */
         $className = $processorData->class;
         $queue = $processorData->queue;
         if ($processorData->type === 'job') {

src/ProcessorData.php

@@ -8,7 +8,7 @@ class ProcessorData
 
     public function __construct(
         public string $class,
-        public ?string $topic = null,
+        public ?string $topicKey = null,
         public ?string $consumer = null,
         public string $type = 'action',
         public string|bool $queue = false,