{consumer|provider}.disconnect() now returns a boolean

src/interface.ts

@@ -3,12 +3,12 @@ import { EachMessageHandler, ProducerRecord } from 'kafkajs';
 export interface Consumer {
   connect(): Promise<void>;
   setCallback(func: EachMessageHandler): void;
-  disconnect(): Promise<void>;
+  disconnect(): Promise<boolean>;
   subscribe(topics: string[], autoCommit: boolean, fromBeginning: boolean): Promise<void>;
 }
 
 export interface Producer {
   send(message: ProducerRecord): Promise<boolean>;
   connect(): Promise<void>;
-  disconnect(): Promise<void>;
+  disconnect(): Promise<boolean>;
 }

src/kafka/kafka-consumer.ts

@@ -46,15 +46,18 @@ export class KafkaConsumer implements GenericConsumer {
     if (this.connected) return;
     await this.consumer.connect();
     this.connected = true;
+    this.logger.info('connected');
   }
 
-  public async disconnect(): Promise<void> {
-    if (!this.connected) return;
+  public async disconnect(): Promise<boolean> {
+    if (!this.connected) return true;
     try {
       await this.consumer.disconnect();
       this.connected = false;
+      return true;
     } catch (err) {
       this.logger.error(`Error on disconnect due to: ${err}`);
     }
+    return false;
   }
 }

src/kafka/kafka-producer.ts

@@ -40,13 +40,15 @@ export class KafkaProducer implements GenericProducer {
     }
   }
 
-  public async disconnect(): Promise<void> {
-    if (!this.connected) return;
+  public async disconnect(): Promise<boolean> {
+    if (!this.connected) return true;
     try {
       await this.producer.disconnect();
       this.connected = false;
+      return true;
     } catch (err) {
       this.logger.error(`Error on disconnect due to: ${err}`);
     }
+    return false;
   }
 }