[upd] Добавлено больше гибкости для логирования

Author: kEERill

config/kafka-consumer.php

@@ -4,7 +4,9 @@
    'global_middleware' => [],
 
    'processors' => [],
-   
+
+   'log_channel' => env('KAFKA_CONSUMER_LOG_CHANNEL', 'null'),
+
    'consumer_options' => [
       /** options for consumer with name `default` */
       'default' => [
@@ -19,6 +21,6 @@
          | Array of middleware.
          */
         'middleware' => [],
-      ]
-   ]
+      ],
+   ],
 ];

src/Commands/KafkaConsumeCommand.php

@@ -5,9 +5,10 @@
 use Ensi\LaravelPhpRdKafka\KafkaFacade;
 use Ensi\LaravelPhpRdKafkaConsumer\ConsumerOptions;
 use Ensi\LaravelPhpRdKafkaConsumer\HighLevelConsumer;
+use Ensi\LaravelPhpRdKafkaConsumer\Logger\ConsumerLoggerFactory;
+use Ensi\LaravelPhpRdKafkaConsumer\Logger\ConsumerLoggerInterface;
 use Ensi\LaravelPhpRdKafkaConsumer\ProcessorData;
 use Illuminate\Console\Command;
-use Illuminate\Support\Facades\Log;
 use Symfony\Component\Console\Command\SignalableCommandInterface;
 use Throwable;
 
@@ -54,29 +55,31 @@ public function handleSignal(int $signal, int|false $previousExitCode = 0): int|
     /**
      * Execute the console command.
      */
-    public function handle(HighLevelConsumer $highLevelConsumer): int
+    public function handle(HighLevelConsumer $highLevelConsumer, ConsumerLoggerFactory $loggerFactory): int
     {
         $this->consumer = $highLevelConsumer;
         $topicKey = $this->argument('topic-key');
         $consumer = $this->argument('consumer');
 
+        $logger = $loggerFactory->make($topicKey, $consumer);
+
         $processorData = $this->findMatchedProcessor($topicKey, $consumer);
         if (is_null($processorData)) {
-            $this->error("Processor for topic-key \"$topicKey\" and consumer \"$consumer\" is not found");
+            $this->errorMessage($logger, "Processor for topic-key \"$topicKey\" and consumer \"$consumer\" is not found");
             $this->line('Processors are set in /config/kafka-consumer.php');
 
             return 1;
         }
 
         if (!class_exists($processorData->class)) {
-            $this->error("Processor class \"$processorData->class\" is not found");
+            $this->errorMessage($logger, "Processor class \"$processorData->class\" is not found");
             $this->line('Processors are set in /config/kafka-consumer.php');
 
             return 1;
         }
 
         if (!$processorData->hasValidType()) {
-            $this->error("Invalid processor type \"$processorData->type\", supported types are: " . implode(',', $processorData->getSupportedTypes()));
+            $this->errorMessage($logger, "Invalid processor type \"$processorData->type\", supported types are: " . implode(',', $processorData->getSupportedTypes()));
 
             return 1;
         }
@@ -97,9 +100,7 @@ public function handle(HighLevelConsumer $highLevelConsumer): int
                 ->for($consumer)
                 ->listen($topicName, $processorData, $consumerOptions);
         } catch (Throwable $e) {
-            Log::error($e->getMessage(), ['exception' => $e]);
-
-            $this->error('An error occurred while listening to the topic: '. $e->getMessage(). ' '. $e->getFile() . '::' . $e->getLine());
+            $this->errorThrowable($logger, $e);
 
             return 1;
         }
@@ -136,4 +137,16 @@ protected function collectMiddleware(array $processorMiddleware): array
             )
         );
     }
+
+    private function errorThrowable(ConsumerLoggerInterface $logger, Throwable $e): void
+    {
+        $logger->error($e->getMessage(), ['exception' => $e]);
+        $this->error('An error occurred while listening to the topic: '. $e->getMessage(). ' '. $e->getFile() . '::' . $e->getLine());
+    }
+
+    private function errorMessage(ConsumerLoggerInterface $logger, string $message): void
+    {
+        $logger->error($message);
+        $this->error($message);
+    }
 }

src/Logger/ConsumerLogger.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace Ensi\LaravelPhpRdKafkaConsumer\Logger;
+
+use Psr\Log\AbstractLogger;
+use Psr\Log\LoggerInterface;
+use Stringable;
+
+final class ConsumerLogger extends AbstractLogger implements ConsumerLoggerInterface
+{
+    public function __construct(
+        protected LoggerInterface $logger,
+        protected string $topicKey,
+        protected string $consumer,
+    ) {
+    }
+
+    public function getTopicKey(): string
+    {
+        return $this->topicKey;
+    }
+
+    public function getConsumer(): string
+    {
+        return $this->consumer;
+    }
+
+    public function getLogger(): LoggerInterface
+    {
+        return $this->logger;
+    }
+
+    public function log($level, Stringable|string $message, array $context = []): void
+    {
+        $context = array_merge($context, [
+            'topic_key' => $this->topicKey,
+            'consumer' => $this->consumer,
+        ]);
+
+        $this->logger->log($level, $message, $context);
+    }
+}

src/Logger/ConsumerLoggerFactory.php

@@ -0,0 +1,30 @@
+<?php
+
+namespace Ensi\LaravelPhpRdKafkaConsumer\Logger;
+
+use Illuminate\Support\Facades\Log;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+
+final class ConsumerLoggerFactory
+{
+    public function make(string $topicKey, string $consumer): ConsumerLoggerInterface
+    {
+        return new ConsumerLogger(
+            $this->makeLogger(),
+            $topicKey,
+            $consumer
+        );
+    }
+
+    private function makeLogger(): LoggerInterface
+    {
+        $channel = config('kafka-consumer.log_channel', 'null');
+
+        if (empty($channel) || $channel == 'null') {
+            return new NullLogger();
+        }
+
+        return Log::channel($channel);
+    }
+}

src/Logger/ConsumerLoggerInterface.php

@@ -0,0 +1,14 @@
+<?php
+
+namespace Ensi\LaravelPhpRdKafkaConsumer\Logger;
+
+use Psr\Log\LoggerInterface;
+
+interface ConsumerLoggerInterface extends LoggerInterface
+{
+    public function getLogger(): LoggerInterface;
+
+    public function getTopicKey(): string;
+
+    public function getConsumer(): string;
+}

tests/ConsumerLoggerFactoryTest.php

@@ -0,0 +1,18 @@
+<?php
+
+use Ensi\LaravelPhpRdKafkaConsumer\Logger\ConsumerLoggerFactory;
+use Psr\Log\NullLogger;
+
+test('test create consumer logger from factory', function (?string $channel) {
+    config()->set('kafka-consumer.log_channel', $channel);
+
+    $consumerLogger = (new ConsumerLoggerFactory())
+        ->make('products', 'default');
+
+    expect($consumerLogger->getLogger())
+        ->toBeInstanceOf(NullLogger::class)
+        ->and($consumerLogger->getTopicKey())
+        ->toBe('products')
+        ->and($consumerLogger->getConsumer())
+        ->toBe('default');
+})->with([null, 'null']);

tests/ConsumerLoggerTest.php

@@ -0,0 +1,22 @@
+<?php
+
+use _PHPStan_01e5828ef\Psr\Log\LogLevel;
+use Ensi\LaravelPhpRdKafkaConsumer\Logger\ConsumerLogger;
+use Psr\Log\NullLogger;
+
+test('merging context when call log', function () {
+    $logger = mock(NullLogger::class)
+        ->makePartial();
+
+    $consumerLogger = new ConsumerLogger($logger, 'products', 'default');
+    $consumerLogger->log(LogLevel::NOTICE, 'Test message', ['foo' => 'bar']);
+
+    $logger->shouldHaveReceived('log', function ($level, $message, array $context = []) {
+        return $level == LogLevel::NOTICE
+            && $message === 'Test message'
+            && count($context) == 3
+            && $context['topic_key'] === 'products'
+            && $context['consumer'] === 'default'
+            && $context['foo'] == 'bar';
+    });
+});