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

* ломающее изменение: теперь kafka:consume и kafka-consumer.processors.topic принимают не название топика, а ключ в kafka.topics
* добавлен пакет kwn/php-rdkafka-stubs

Author: Ivan Koryukov

.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": "dev-task-102492",
         "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,7 +54,7 @@ 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', []));
 
@@ -64,9 +65,9 @@ public function handle(HighLevelConsumer $highLevelConsumer): int
             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 +94,13 @@ public function handle(HighLevelConsumer $highLevelConsumer): int
             middleware: $this->collectMiddleware($consumerPackageOptions['middleware'] ?? []),
         );
 
-        $this->info("Start listenning to topic: \"$topic\", consumer \"$consumer\"");
+        $topicName = KafkaFacade::topicName($topicKey);
+        $this->info("Start listenning to topic: \"$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());
 
@@ -117,7 +119,7 @@ protected function findMatchedProcessor(string $topic, string $consumer): ?Proce
                 ) {
                 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,