Merge pull request #12954 from gunb0s/feature/nestjs-kafka-emit-batch

feat: emit batch

packages/microservices/client/client-kafka.ts

@@ -1,6 +1,6 @@
 import { Logger } from '@nestjs/common/services/logger.service';
 import { loadPackage } from '@nestjs/common/utils/load-package.util';
-import { isUndefined } from '@nestjs/common/utils/shared.utils';
+import { isNil, isUndefined } from '@nestjs/common/utils/shared.utils';
 import {
   KAFKA_DEFAULT_BROKER,
   KAFKA_DEFAULT_CLIENT,
@@ -37,6 +37,15 @@ import {
   KafkaRequestSerializer,
 } from '../serializers/kafka-request.serializer';
 import { ClientProxy } from './client-proxy';
+import {
+  connectable,
+  defer,
+  Observable,
+  Subject,
+  throwError as _throw,
+} from 'rxjs';
+import { mergeMap } from 'rxjs/operators';
+import { InvalidMessageException } from '../errors/invalid-message.exception';
 
 let kafkaPackage: any = {};
 
@@ -223,6 +232,58 @@ export class ClientKafka extends ClientProxy {
     return this.consumerAssignments;
   }
 
+  public emitBatch<TResult = any, TInput = any>(
+    pattern: any,
+    data: { messages: TInput[] },
+  ): Observable<TResult> {
+    if (isNil(pattern) || isNil(data)) {
+      return _throw(() => new InvalidMessageException());
+    }
+    const source = defer(async () => this.connect()).pipe(
+      mergeMap(() => this.dispatchBatchEvent({ pattern, data })),
+    );
+    const connectableSource = connectable(source, {
+      connector: () => new Subject(),
+      resetOnDisconnect: false,
+    });
+    connectableSource.connect();
+    return connectableSource;
+  }
+
+  public commitOffsets(
+    topicPartitions: TopicPartitionOffsetAndMetadata[],
+  ): Promise<void> {
+    if (this.consumer) {
+      return this.consumer.commitOffsets(topicPartitions);
+    } else {
+      throw new Error('No consumer initialized');
+    }
+  }
+
+  protected async dispatchBatchEvent<TInput = any>(
+    packets: ReadPacket<{ messages: TInput[] }>,
+  ): Promise<any> {
+    if (packets.data.messages.length === 0) {
+      return;
+    }
+    const pattern = this.normalizePattern(packets.pattern);
+    const outgoingEvents = await Promise.all(
+      packets.data.messages.map(message => {
+        return this.serializer.serialize(message as any, { pattern });
+      }),
+    );
+
+    const message = Object.assign(
+      {
+        topic: pattern,
+        messages: outgoingEvents,
+      },
+      this.options.send || {},
+    );
+
+    return this.producer.send(message);
+  }
+
   protected async dispatchEvent(packet: OutgoingEvent): Promise<any> {
     const pattern = this.normalizePattern(packet.pattern);
     const outgoingEvent = await this.serializer.serialize(packet.data, {
@@ -320,14 +381,4 @@ export class ClientKafka extends ClientProxy {
     this.deserializer =
       (options && options.deserializer) || new KafkaResponseDeserializer();
   }
-
-  public commitOffsets(
-    topicPartitions: TopicPartitionOffsetAndMetadata[],
-  ): Promise<void> {
-    if (this.consumer) {
-      return this.consumer.commitOffsets(topicPartitions);
-    } else {
-      throw new Error('No consumer initialized');
-    }
-  }
 }

packages/microservices/test/client/client-kafka.spec.ts

@@ -9,6 +9,8 @@ import {
   EachMessagePayload,
   KafkaMessage,
 } from '../../external/kafka.interface';
+import { Observable } from 'rxjs';
+import { Producer } from 'kafkajs';
 
 describe('ClientKafka', () => {
   const topic = 'test.topic';
@@ -549,6 +551,77 @@ describe('ClientKafka', () => {
     });
   });
 
+  describe('emitBatch', () => {
+    it(`should return an observable stream`, () => {
+      const stream$ = client.emitBatch(
+        {},
+        {
+          messages: [],
+        },
+      );
+      expect(stream$ instanceof Observable).to.be.true;
+    });
+
+    it(`should call "connect" immediately`, () => {
+      const connectSpy = sinon.spy(client, 'connect');
+      client.emitBatch(
+        {},
+        {
+          messages: [],
+        },
+      );
+      expect(connectSpy.calledOnce).to.be.true;
+    });
+
+    describe('when "connect" throws', () => {
+      it('should return Observable with error', () => {
+        sinon.stub(client, 'connect').callsFake(() => {
+          throw new Error();
+        });
+
+        const stream$ = client.emitBatch(
+          {},
+          {
+            messages: [],
+          },
+        );
+
+        stream$.subscribe({
+          next: () => {},
+          error: err => {
+            expect(err).to.be.instanceof(Error);
+          },
+        });
+      });
+    });
+
+    describe('when is connected', () => {
+      beforeEach(() => {
+        sinon
+          .stub(client, 'connect')
+          .callsFake(() => Promise.resolve({} as Producer));
+      });
+
+      it(`should call dispatchBatchEvent`, () => {
+        const pattern = { test: 3 };
+        const data = { messages: [] };
+        const dispatchBatchEventSpy = sinon
+          .stub()
+          .callsFake(() => Promise.resolve(true));
+        const stream$ = client.emitBatch(pattern, data);
+        client['dispatchBatchEvent'] = dispatchBatchEventSpy;
+        stream$.subscribe(() => {
+          expect(dispatchBatchEventSpy.calledOnce).to.be.true;
+        });
+      });
+    });
+
+    it('should return Observable with error', () => {
+      const err$ = client.emitBatch(null, null);
+      expect(err$).to.be.instanceOf(Observable);
+    });
+  });
+
   describe('dispatchEvent', () => {
     const eventMessage = {
       id: undefined,