add tests for bindEvent empty messageHandlers.

integration/microservices/e2e/sum-kafka.spec.ts

@@ -18,17 +18,17 @@ import { UserDto } from '../src/kafka/dtos/user.dto';
 import { UserEntity } from '../src/kafka/entities/user.entity';
 import { BusinessDto } from '../src/kafka/dtos/business.dto';
 import { BusinessEntity } from '../src/kafka/entities/business.entity';
+import * as async from 'async';
 
 @Catch()
 class KafkaExceptionFilter implements ExceptionFilter {
   catch(exception: HttpException, host: ArgumentsHost): any {
-    console.log(exception);
+    throw exception;
   }
 }
 @Catch()
 class RpcErrorFilter implements RpcExceptionFilter {
   catch(exception: RpcException): Observable<any> {
-    console.log(exception);
     return throwError(exception);
   }
 }
@@ -108,13 +108,27 @@ describe('Kafka transport', () => {
   }).timeout(50000);
 
   it(`/POST (async command create business`, () => {
-    const newBusiness: BusinessEntity = new BusinessEntity(businessDto);
     return request(server)
       .post('/business')
       .send(businessDto)
       .expect(200);
   }).timeout(50000);
 
+  it(`/POST (async command create user) Concurrency Test`, (done) => {
+    const tasks = [];
+    for (let concurrencyKey = 0; concurrencyKey < 100; concurrencyKey++) {
+      tasks.push((cb) => {
+        const innerUserDto = JSON.parse(JSON.stringify(userDto));
+        innerUserDto.name += `+${concurrencyKey}`;
+        request(server)
+          .post('/user')
+          .send(userDto)
+          .expect(200, cb);
+      });
+    }
+    async.parallel(tasks, done);
+  }).timeout(50000);
+
   after(`Stopping Kafka app`, async () => {
     await app.close();
   });

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

@@ -1,9 +1,13 @@
 import { expect } from 'chai';
 import * as sinon from 'sinon';
-import { NO_MESSAGE_HANDLER } from '../../constants';
-import { ServerTCP } from '../../server/server-tcp';
 import { ServerKafka } from '../../server';
-import { Consumer, IHeaders, KafkaMessage } from '../../external/kafka.interface';
+import { Logger } from '@nestjs/common';
+
+class NoopLogger extends Logger {
+  log(message: any, context?: string): void {}
+  error(message: any, trace?: string, context?: string): void {}
+  warn(message: any, context?: string): void {}
+}
 
 describe('ServerKafka', () => {
   let server: ServerKafka;
@@ -12,9 +16,6 @@ describe('ServerKafka', () => {
     server = new ServerKafka({});
   });
 
-  afterEach(() => {
-    server.close();
-  });
   describe('close', () => {
     it('should close server', () => {
       server.close();
@@ -23,17 +24,67 @@ describe('ServerKafka', () => {
       expect(server.client).to.be.null;
     });
   });
+
+  let callback: sinon.SinonSpy;
+  let bindEventsStub: sinon.SinonStub;
+  let connect: sinon.SinonSpy;
+  let subscribe: sinon.SinonSpy;
+  let run: sinon.SinonSpy;
+  let consumerStub: sinon.SinonStub;
+  let producerStub: sinon.SinonStub;
+  let client;
+  beforeEach(() => {
+    callback = sinon.spy();
+    connect = sinon.spy();
+    subscribe = sinon.spy();
+    run = sinon.spy();
+    bindEventsStub = sinon
+      .stub(server, 'bindEvents')
+      .callsFake(() => ({} as any));
+    consumerStub = sinon.stub(server, 'consumer')
+      .callsFake( () => {
+        return {
+          connect,
+          subscribe,
+          run,
+        };
+      });
+    producerStub = sinon.stub(server, 'producer')
+      .callsFake( () => {
+        return {
+          connect,
+        };
+      });
+    client = {
+      consumer: consumerStub,
+      producer: producerStub,
+    };
+    sinon.stub(server, 'createClient').callsFake(() => client);
+  });
+
   describe('listen', () => {
-    it('should start server', async () => {
-      const callback = sinon.spy();
+    it('should call "bindEvents"', async () => {
+      await server.listen(callback);
+      expect(bindEventsStub.called).to.be.true;
+    });
+    it('should call "client.start"', async () => {
+
+      await server.listen(callback);
+      expect(client.producer.called).to.be.true;
+    });
+    it('should call callback', async () => {
       await server.listen(callback);
       expect(callback.called).to.be.true;
     });
-    it('should have kafka, producer and consumer connected', async () => {
-      await server.listen(() => {});
-      expect(server.client).to.be.not.null;
-      expect(server.producer).to.be.not.null;
-      expect(server.consumer).to.be.not.null;
+  });
+
+  describe('bindEvents', () => {
+    it('should not call subscribe nor run on consumer when there are no messageHandlers', async () => {
+      (server as any).logger = new NoopLogger();
+      await server.bindEvents(server.consumer);
+      expect(subscribe.called).to.be.not.true;
+      expect(run.called).to.be.not.true;
+      expect(connect.called).to.be.not.true;
     });
   });
 });