Feat: kafka is connected (#329)

* Updating deps

* Kafka consumer exposing isConnected and isActive

* Release prepare

Author: CarlosGamero

packages/kafka/lib/AbstractKafkaConsumer.ts

@@ -111,6 +111,21 @@ export abstract class AbstractKafkaConsumer<
     /* v8 ignore stop */
   }
 
+  /**
+   * Returns true if all client's connections are currently connected and the client is connected to at least one broker.
+   */
+  get isConnected(): boolean {
+    return this.consumer.isConnected()
+  }
+
+  /**
+   * Returns `true` if the consumer is not closed, and it is currently an active member of a consumer group.
+   * This method will return `false` during consumer group rebalancing.
+   */
+  get isActive(): boolean {
+    return this.consumer.isActive()
+  }
+
   async init(): Promise<void> {
     if (this.consumerStream) return Promise.resolve()
     const topics = this.handlerContainer.topics

packages/kafka/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@message-queue-toolkit/kafka",
-    "version": "0.5.1",
+    "version": "0.6.0",
     "engines": {
         "node": ">= 22.14.0"
     },
@@ -49,7 +49,7 @@
     "dependencies": {
         "@lokalise/node-core": "^14.2.0",
         "@lokalise/universal-ts-utils": "^4.5.1",
-        "@platformatic/kafka": "^1.9.0"
+        "@platformatic/kafka": "^1.10.5"
     },
     "peerDependencies": {
         "@message-queue-toolkit/core": ">=23.0.0",

packages/kafka/test/consumer/PermissionConsumer.spec.ts

@@ -114,6 +114,52 @@ describe('PermissionConsumer', () => {
     })
   })
 
+  describe('isConnected', () => {
+    it('should return false if consumer is not initiated', () => {
+      // Given
+      consumer = new PermissionConsumer(testContext.cradle)
+
+      // When - Then
+      expect(consumer.isConnected).toBe(false)
+    })
+
+    it('should return true if consumer is initiated', async () => {
+      // Given
+      consumer = new PermissionConsumer(testContext.cradle)
+
+      // When
+      await consumer.init()
+
+      // Then
+      expect(consumer.isConnected).toBe(true)
+      await consumer.close()
+      expect(consumer.isConnected).toBe(false)
+    })
+  })
+
+  describe('isActive', () => {
+    it('should return false if consumer is not initiated', () => {
+      // Given
+      consumer = new PermissionConsumer(testContext.cradle)
+
+      // When - Then
+      expect(consumer.isActive).toBe(false)
+    })
+
+    it('should return true if consumer is initiated', async () => {
+      // Given
+      consumer = new PermissionConsumer(testContext.cradle)
+
+      // When
+      await consumer.init()
+
+      // Then
+      expect(consumer.isActive).toBe(true)
+      await consumer.close()
+      expect(consumer.isActive).toBe(false)
+    })
+  })
+
   describe('consume', () => {
     let publisher: PermissionPublisher