[add] Добавлены недостающие тесты по созданию Consumer

Author: popkovkirill

src/Commands/KafkaConsumeCommand.php

@@ -84,15 +84,14 @@ public function handle(ConsumerFactory $consumerFactory): int
     {
         try {
             $this->consumer = $consumerFactory
-                ->build($this->getTopicKey(), $this->getConsumer());
+                ->build($this->getTopicKey(), $this->getConsumer())
+                ->setMaxEvents($this->getMaxEvents())
+                ->setMaxTime($this->getMaxTime());
 
             $this->info("Start listening to topic: \"{$this->getTopicKey()}\"" .
                 " ({$this->consumer->getTopicName()}), consumer \"{$this->getConsumer()}\"");
 
-            $this->consumer
-                ->setMaxEvents($this->getMaxEvents())
-                ->setMaxTime($this->getMaxTime())
-                ->listen();
+            $this->consumer->listen();
         } catch (Throwable $exception) {
             $this->errorThrowable($exception);
 

src/Consumers/Consumer.php

@@ -42,6 +42,22 @@ public function forceStop(): void
         $this->highLevelConsumer->forceStop();
     }
 
+    /**
+     * @return ProcessorData
+     */
+    public function getProcessorData(): ProcessorData
+    {
+        return $this->processorData;
+    }
+
+    /**
+     * @return ConsumerOptions
+     */
+    public function getConsumerOptions(): ConsumerOptions
+    {
+        return $this->consumerOptions;
+    }
+
     /**
      * @return void
      *

src/Consumers/Factories/ConsumerFactory.php

@@ -99,11 +99,10 @@ protected function makeConsumerOptions(string $consumer, ProcessorData $processo
 
     protected function collectMiddleware(array $processorMiddleware): array
     {
-        return array_unique(
-            array_merge(
-                config('kafka-consumer.global_middleware', []),
-                $processorMiddleware
-            )
-        );
+        return collect(config('kafka-consumer.global_middleware', []))
+            ->merge($processorMiddleware)
+            ->unique()
+            ->values()
+            ->toArray();
     }
 }

tests/ConsumerFakerTest.php

@@ -1,5 +1,6 @@
 <?php
 
+use Ensi\LaravelPhpRdKafka\KafkaManager;
 use Ensi\LaravelPhpRdKafkaConsumer\Tests\ConsumerFaker;
 use Ensi\LaravelPhpRdKafkaConsumer\Tests\TestConsumer;
 use RdKafka\Message;
@@ -13,3 +14,14 @@
 
     TestConsumer::assertMessageConsumed($message);
 });
+
+test('bind testing kafka manager with faker consumer', function () {
+    TestConsumer::fake('test-model');
+
+    ConsumerFaker::new('test-model')
+        ->addMessage(new Message())
+        ->bind();
+
+    expect(resolve(KafkaManager::class))
+        ->toBeInstanceOf(\Ensi\LaravelPhpRdKafkaConsumer\Tests\KafkaManager::class);
+});

tests/Consumers/ConsumerTest.php

@@ -33,9 +33,9 @@ class: TestConsumer::class,
 
     $consumer->listen();
 
-    expect($consumerOptions->maxEvents)
+    expect($consumer->getConsumerOptions()->maxEvents)
         ->toBe(10000)
-        ->and($consumerOptions->maxTime)
+        ->and($consumer->getConsumerOptions()->maxTime)
         ->toBe(5100);
 
     $highLevelConsumer->shouldHaveReceived('for', ['default']);

tests/Consumers/Factories/ConsumerFactoryTest.php

@@ -23,3 +23,26 @@
     (new ConsumerFactory(resolve(HighLevelConsumer::class)))
         ->build('test-models');
 })->throws(KafkaConsumerProcessorException::class, 'Invalid processor type "invalid-type", supported types are: action,job');
+
+test('set consume_timeout and middleware to consumer options', function () {
+    config()
+        ->set('kafka-consumer.consumer_options.default', [
+            'consume_timeout' => 55000,
+            'middleware' => ['AlreadyAddedMiddleware', 'KafkaConsumerMiddleware'],
+        ]);
+
+    config()
+        ->set('kafka-consumer.global_middleware', ['AlreadyAddedMiddleware']);
+
+    setConsumerTopicConfig('test-models', TestConsumer::class);
+
+    $consumer = (new ConsumerFactory(resolve(HighLevelConsumer::class)))
+        ->build('test-models');
+
+    $consumerOptions = $consumer->getConsumerOptions();
+
+    expect($consumerOptions->consumeTimeout)
+        ->toBe(55000)
+        ->and($consumerOptions->middleware)
+        ->toBe(['AlreadyAddedMiddleware', 'KafkaConsumerMiddleware']);
+});