Merge pull request #7 from ensi-platform/task-93909-1

#93909 create topic before subscribing

Author: MadridianFox

README.md

@@ -73,6 +73,10 @@ Processors in config have the following configuration options:
 ]
 
 ```
+
+**Important!** Some topics have to have different consumer settings, such as start reading topic from the beginning or don't create topic if it is not exists yet.  
+For such cases you need to configure several consumers and use suitable one.
+
 ### Synchronous processors
 
 Most of the time all tou need is a synchronous processor.

src/HighLevelConsumer.php

@@ -46,7 +46,7 @@ public function forceStop(): static
      */
     public function listen(string $topicName, ProcessorData $processorData, ConsumerOptions $options): void
     {
-        $this->consumer->subscribe([ $topicName ]);
+        $this->subscribe($topicName);
 
         [$startTime, $eventsProcessed] = [hrtime(true) / 1e9, 0];
 
@@ -128,6 +128,21 @@ protected function shouldBeStopped(int|float $startTime, int $eventsProcessed, C
         return $this->forceStop;
     }
 
+    protected function subscribe(string $topicName): void
+    {
+        $attempts = 0;
+        do {
+            $topicExists = (bool)$this->getPartitions($topicName);
+            if (!$topicExists) {
+                $this->consumer->newTopic($topicName);
+                sleep(1);
+            }
+            $attempts++;
+        } while (!$topicExists && $attempts < 10);
+
+        $this->consumer->subscribe([$topicName]);
+    }
+
     /**
      * @param string $topicName
      * @return TopicPartition[]