feat: adding support to receive kafka message headers on topic subscr…

…iber method

README.md

@@ -107,7 +107,7 @@ export class Consumer {
   }
 
   @SubscribeTo('hero.kill.dragon')
-  async getWorld(data: any, key: any, offset: number, timestamp: number): Promise<void> {
+  async getWorld(data: any, key: any, offset: number, timestamp: number, partition: number, headers: IHeaders): Promise<void> {
     ...
   }
 

src/deserializer/kafka-response.deserializer.ts

@@ -1,27 +1,34 @@
-import { Deserializer } from "@nestjs/microservices";
-import { KafkaResponse } from "../interfaces";
+import { Deserializer } from '@nestjs/microservices';
+import { KafkaResponse } from '../interfaces';
 
 export class KafkaResponseDeserializer
-  implements Deserializer<any, KafkaResponse> {
+  implements Deserializer<any, KafkaResponse>
+{
+  deserialize(message: any, options?: Record<string, any>): KafkaResponse {
+    const { key, value, timestamp, offset, headers } = message;
+    let id = key;
+    let response = value;
 
-    deserialize(message: any, options?: Record<string, any>): KafkaResponse {
-      const { key, value, timestamp, offset } = message;
-      let id = key;
-      let response = value;
+    if (Buffer.isBuffer(key)) {
+      id = Buffer.from(key).toString();
+    }
 
-      if (Buffer.isBuffer(key)) {
-        id = Buffer.from(key).toString()
-      }
+    if (Buffer.isBuffer(value)) {
+      response = Buffer.from(value).toString();
+    }
 
-      if (Buffer.isBuffer(value)) {
-        response = Buffer.from(value).toString()
+    Object.keys(headers).forEach((key) => {
+      if (Buffer.isBuffer(headers[key])) {
+        headers[key] = Buffer.from(headers[key]).toString();
       }
+    });
 
-      return {
-        key: id,
-        response,
-        timestamp,
-        offset
-      };
-    }
+    return {
+      key: id,
+      response,
+      timestamp,
+      offset,
+      headers,
+    };
+  }
 }

src/interfaces.ts

@@ -11,11 +11,15 @@ import {
 } from 'kafkajs';
 import { ModuleMetadata, Type } from '@nestjs/common';
 
+export interface IHeaders {
+  [key: string]: any;
+}
 export interface KafkaResponse<T = any> {
   response: T;
   key: string;
   timestamp: string;
   offset: number;
+  headers?: IHeaders;
 }
 export interface KafkaModuleOption {
   name: string;
@@ -28,7 +32,7 @@ export interface KafkaModuleOption {
     serializer?: Serializer;
     consumeFromBeginning?: boolean;
     seek?: Record<string, number | 'earliest' | Date>;
-    autoConnect?: boolean
+    autoConnect?: boolean;
   };
 }
 export interface KafkaMessageObject extends Message {
@@ -51,6 +55,7 @@ export interface KafkaModuleOptionsAsync
 export interface KafkaOptionsFactory {
   creatKafkaModuleOptions(): Promise<KafkaModuleOption[]> | KafkaModuleOption[];
 }
-export interface KafkaTransaction extends Omit<Transaction, 'send' | 'sendBatch'> {
-  send(message: KafkaMessageSend): Promise<RecordMetadata[]>
-}
+export interface KafkaTransaction
+  extends Omit<Transaction, 'send' | 'sendBatch'> {
+  send(message: KafkaMessageSend): Promise<RecordMetadata[]>;
+}

src/kafka.service.ts

@@ -245,8 +245,8 @@ export class KafkaService implements OnModuleInit, OnModuleDestroy {
         const callback = SUBSCRIBER_MAP.get(topic);
 
         try {
-          const { timestamp, response, offset, key } = await this.deserializer.deserialize(message, { topic });
-          await callback.apply(objectRef, [response, key, offset, timestamp, partition]);
+          const { timestamp, response, offset, key, headers } = await this.deserializer.deserialize(message, { topic });
+          await callback.apply(objectRef, [response, key, offset, timestamp, partition, headers]);
         } catch(e) {
           this.logger.error(`Error for message ${topic}: ${e}`);