[upd] Переименован параметр $consumer в $consumerName, как это реализовано в пакете laravel-rdkafka-producer

Author: popkovkirill

src/Commands/KafkaConsumeCommand.php

@@ -52,7 +52,7 @@ public function getTopicKey(): string
         return $this->argument('topic-key');
     }
 
-    public function getConsumer(): string
+    public function getConsumerName(): string
     {
         return $this->argument('consumer');
     }
@@ -84,12 +84,12 @@ public function handle(ConsumerFactory $consumerFactory): int
     {
         try {
             $this->consumer = $consumerFactory
-                ->build($this->getTopicKey(), $this->getConsumer())
+                ->build($this->getTopicKey(), $this->getConsumerName())
                 ->setMaxEvents($this->getMaxEvents())
                 ->setMaxTime($this->getMaxTime());
 
             $this->info("Start listening to topic: \"{$this->getTopicKey()}\"" .
-                " ({$this->consumer->getTopicName()}), consumer \"{$this->getConsumer()}\"");
+                " ({$this->consumer->getTopicName()}), consumer \"{$this->getConsumerName()}\"");
 
             $this->consumer->listen();
         } catch (Throwable $exception) {
@@ -123,6 +123,6 @@ private function errorThrowable(Throwable $exception): void
     private function makeLogger(): ConsumerLoggerInterface
     {
         return $this->loggerFactory
-            ->make($this->getTopicKey(), $this->getConsumer());
+            ->make($this->getTopicKey(), $this->getConsumerName());
     }
 }

src/Consumers/Factories/ConsumerFactory.php

@@ -18,34 +18,34 @@ public function __construct(
 
     /**
      * @param string $topicKey
-     * @param string $consumer
+     * @param string $consumerName
      * @return Consumer
      *
      * @throws KafkaConsumerProcessorException
      */
-    public function build(string $topicKey, string $consumer = 'default'): Consumer
+    public function build(string $topicKey, string $consumerName = 'default'): Consumer
     {
-        $processorData = $this->makeProcessorData($topicKey, $consumer);
-        $consumerOptions = $this->makeConsumerOptions($consumer, $processorData);
+        $processorData = $this->makeProcessorData($topicKey, $consumerName);
+        $consumerOptions = $this->makeConsumerOptions($consumerName, $processorData);
 
         return new Consumer(
             highLevelConsumer: $this->highLevelConsumer,
             processorData: $processorData,
             consumerOptions: $consumerOptions,
-            topicName: KafkaFacade::topicNameByClient('consumer', $consumer, $topicKey)
+            topicName: KafkaFacade::topicNameByClient('consumer', $consumerName, $topicKey)
         );
     }
 
     /**
      * @param string $topicKey
-     * @param string $consumer
+     * @param string $consumerName
      * @return ProcessorData
      *
      * @throws KafkaConsumerProcessorException
      */
-    protected function makeProcessorData(string $topicKey, string $consumer): ProcessorData
+    protected function makeProcessorData(string $topicKey, string $consumerName): ProcessorData
     {
-        $processorData = $this->findMatchedProcessor($topicKey, $consumer);
+        $processorData = $this->findMatchedProcessor($topicKey, $consumerName);
 
         if (!class_exists($processorData->class)) {
             throw new KafkaConsumerProcessorException("Processor class \"$processorData->class\" is not found");
@@ -61,35 +61,35 @@ protected function makeProcessorData(string $topicKey, string $consumer): Proces
 
     /**
      * @param string $topicKey
-     * @param string $consumer
+     * @param string $consumerName
      * @return ProcessorData
      *
      * @throws KafkaConsumerProcessorException
      */
-    protected function findMatchedProcessor(string $topicKey, string $consumer): ProcessorData
+    protected function findMatchedProcessor(string $topicKey, string $consumerName): ProcessorData
     {
         foreach (config('kafka-consumer.processors', []) as $processor) {
             $topicMatched = empty($processor['topic']) || $processor['topic'] === $topicKey;
-            $consumerMatched = empty($processor['consumer']) || $processor['consumer'] === $consumer;
+            $consumerMatched = empty($processor['consumer']) || $processor['consumer'] === $consumerName;
 
             if ($topicMatched && $consumerMatched) {
                 return new ProcessorData(
                     class: $processor['class'],
                     topicKey: $processor['topic'] ?? $topicKey,
-                    consumer: $processor['consumer'] ?? $consumer,
+                    consumer: $processor['consumer'] ?? $consumerName,
                     type: $processor['type'] ?? 'action',
                     queue: $processor['queue'] ?? false,
                     consumeTimeout: $processor['consume_timeout'] ?? 20000,
                 );
             }
         }
 
-        throw new KafkaConsumerProcessorException("Processor for topic-key \"$topicKey\" and consumer \"$consumer\" is not found");
+        throw new KafkaConsumerProcessorException("Processor for topic-key \"$topicKey\" and consumer \"$consumerName\" is not found");
     }
 
-    protected function makeConsumerOptions(string $consumer, ProcessorData $processorData): ConsumerOptions
+    protected function makeConsumerOptions(string $consumerName, ProcessorData $processorData): ConsumerOptions
     {
-        $consumerPackageOptions = config('kafka-consumer.consumer_options.' . $consumer, []);
+        $consumerPackageOptions = config('kafka-consumer.consumer_options.' . $consumerName, []);
 
         return new ConsumerOptions(
             consumeTimeout: $consumerPackageOptions['consume_timeout'] ?? $processorData->consumeTimeout,

src/Loggers/ConsumerLogger.php

@@ -10,8 +10,8 @@ final class ConsumerLogger extends AbstractLogger implements ConsumerLoggerInter
 {
     public function __construct(
         protected LoggerInterface $logger,
-        protected string $topicKey,
-        protected string $consumer,
+        protected string          $topicKey,
+        protected string          $consumerName,
     ) {
     }
 
@@ -20,9 +20,9 @@ public function getTopicKey(): string
         return $this->topicKey;
     }
 
-    public function getConsumer(): string
+    public function getConsumerName(): string
     {
-        return $this->consumer;
+        return $this->consumerName;
     }
 
     public function getLogger(): LoggerInterface
@@ -34,7 +34,7 @@ public function log($level, Stringable|string $message, array $context = []): vo
     {
         $context = array_merge($context, [
             'topic_key' => $this->topicKey,
-            'consumer' => $this->consumer,
+            'consumer_name' => $this->consumerName,
         ]);
 
         $this->logger->log($level, $message, $context);

src/Loggers/ConsumerLoggerFactory.php

@@ -8,12 +8,12 @@
 
 final class ConsumerLoggerFactory
 {
-    public function make(string $topicKey, string $consumer): ConsumerLoggerInterface
+    public function make(string $topicKey, string $consumerName = 'default'): ConsumerLoggerInterface
     {
         return new ConsumerLogger(
             $this->makeLogger(),
             $topicKey,
-            $consumer
+            $consumerName
         );
     }
 

src/Loggers/ConsumerLoggerInterface.php

@@ -10,5 +10,5 @@ public function getLogger(): LoggerInterface;
 
     public function getTopicKey(): string;
 
-    public function getConsumer(): string;
+    public function getConsumerName(): string;
 }

src/Tests/ConsumerFaker.php

@@ -20,9 +20,9 @@ class ConsumerFaker
 
     public function __construct(
         protected string $topicKey,
-        protected string $consumer = 'default'
+        protected string $consumerName = 'default'
     ) {
-        $this->topicName = KafkaFacade::topicNameByClient('consumer', $consumer, $topicKey);
+        $this->topicName = KafkaFacade::topicNameByClient('consumer', $consumerName, $topicKey);
     }
 
     public function addMessage(Message $message): self
@@ -50,7 +50,7 @@ public function consume(): void
         $this->bind();
 
         (new ConsumerFactory(resolve(HighLevelConsumer::class)))
-            ->build($this->topicKey, $this->consumer)
+            ->build($this->topicKey, $this->consumerName)
             ->listen();
     }
 
@@ -71,8 +71,8 @@ private function makeKafkaManager(): KafkaManager
         return new KafkaManager(new KafkaConsumer($this->topicName, $this->messages));
     }
 
-    public static function new(string $topicKey, string $consumer = 'default'): self
+    public static function new(string $topicKey, string $consumerName = 'default'): self
     {
-        return new self($topicKey, $consumer);
+        return new self($topicKey, $consumerName);
     }
 }

tests/Loggers/ConsumerLoggerFactoryTest.php

@@ -13,6 +13,6 @@
         ->toBeInstanceOf(NullLogger::class)
         ->and($consumerLogger->getTopicKey())
         ->toBe('products')
-        ->and($consumerLogger->getConsumer())
+        ->and($consumerLogger->getConsumerName())
         ->toBe('default');
 })->with([null, 'null']);

tests/Loggers/ConsumerLoggerTest.php

@@ -16,7 +16,7 @@
             && $message === 'Test message'
             && count($context) == 3
             && $context['topic_key'] === 'products'
-            && $context['consumer'] === 'default'
+            && $context['consumer_name'] === 'default'
             && $context['foo'] == 'bar';
     });
 });