Revert "fix: use onApplicationBootstrap to start processing messages"

This reverts commit b4b890c.

src/consumer/kafkaConsumer.ts

@@ -18,7 +18,7 @@
 
 import vm from "vm";
 
-import { Inject, Injectable, OnApplicationBootstrap } from "@nestjs/common";
+import { Inject, Injectable } from "@nestjs/common";
 import {
   CustomTransportStrategy,
   MessageHandler,
@@ -37,10 +37,7 @@ import { IKafkaConsumerSerializedOptions } from "./interfaces";
 import { KafkaConsumerMessageHandler } from "./kafkaConsumerMessageHandler";
 
 @Injectable()
-export class KafkaConsumer
-  extends Server
-  implements CustomTransportStrategy, OnApplicationBootstrap
-{
+export class KafkaConsumer extends Server implements CustomTransportStrategy {
   public readonly transportId = KafkaConsumerTransportId;
 
   private readonly messageHandlersMap: Map<
@@ -57,18 +54,15 @@ export class KafkaConsumer
   ) {
     super();
   }
+
   public async close(): Promise<void> {
     await Promise.all(
       [...this.consumersMap.values()].map((x) => x.disconnect()),
     );
   }
 
   public listen(callback: (error?: unknown) => void): void {
-    this.connectAndSubscribe().then(callback).catch(callback);
-  }
-
-  public async onApplicationBootstrap(): Promise<void> {
-    await this.startProcessingMessages();
+    this.start().then(callback).catch(callback);
   }
 
   private addMessageHandlerToMap(
@@ -87,12 +81,6 @@ export class KafkaConsumer
     }
   }
 
-  private async connectAndSubscribe(): Promise<void> {
-    await Promise.all([...this.consumersMap.values()].map((x) => x.connect()));
-
-    await this.subscribe();
-  }
-
   private getConsumer(connectionName: string): Consumer {
     const consumer = this.consumersMap.get(connectionName);
 
@@ -105,7 +93,7 @@ export class KafkaConsumer
     return consumer;
   }
 
-  private async startProcessingMessages(): Promise<void> {
+  private async run(): Promise<void> {
     for (const [connectionName, consumer] of this.consumersMap) {
       await consumer.run({
         eachMessage: async (rawPayload) => {
@@ -123,6 +111,13 @@ export class KafkaConsumer
     }
   }
 
+  private async start(): Promise<void> {
+    await Promise.all([...this.consumersMap.values()].map((x) => x.connect()));
+
+    await this.subscribe();
+    await this.run();
+  }
+
   private async subscribe(): Promise<void> {
     const context = vm.createContext({
       topicPickerArgs: this.kafkaOptions.topicPickerArgs,