feat(microservices): add kafka request deserializer, unwrap request

Author: kamilmysliwiec

packages/microservices/deserializers/kafka-request.deserializer.ts

@@ -0,0 +1,21 @@
+import { IncomingEvent, IncomingRequest } from '../interfaces';
+import { KafkaRequest } from '../serializers/kafka-request.serializer';
+import { IncomingRequestDeserializer } from './incoming-request.deserializer';
+
+export class KafkaRequestDeserializer extends IncomingRequestDeserializer {
+  mapToSchema(
+    data: KafkaRequest,
+    options?: Record<string, any>,
+  ): IncomingRequest | IncomingEvent {
+    if (!options) {
+      return {
+        pattern: undefined,
+        data: undefined,
+      };
+    }
+    return {
+      pattern: options.channel,
+      data: data?.value ?? data,
+    };
+  }
+}

packages/microservices/server/server-kafka.ts

@@ -8,6 +8,7 @@ import {
   NO_MESSAGE_HANDLER,
 } from '../constants';
 import { KafkaContext } from '../ctx-host';
+import { KafkaRequestDeserializer } from '../deserializers/kafka-request.deserializer';
 import { KafkaHeaders, Transport } from '../enums';
 import { KafkaRetriableException } from '../exceptions';
 import {
@@ -184,7 +185,6 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
       });
     }
 
-    // @TODO pass packet.data.value (overide deserializer)
     const response$ = this.transformToObservable(
       await handler(packet.data, kafkaContext),
     );
@@ -276,4 +276,8 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
     this.serializer =
       (options && options.serializer) || new KafkaRequestSerializer();
   }
+
+  protected initializeDeserializer(options: KafkaOptions['options']) {
+    this.deserializer = options?.deserializer ?? new KafkaRequestDeserializer();
+  }
 }