add tests for @MessageRequests for complex data types.

Author: Hexenon

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

@@ -1,17 +1,58 @@
-import { INestApplication } from '@nestjs/common';
-import { Transport } from '@nestjs/microservices';
+import {
+  ArgumentsHost,
+  Catch,
+  ExceptionFilter,
+  HttpException,
+  INestApplication,
+  RpcExceptionFilter,
+} from '@nestjs/common';
+import { RpcException, Transport } from '@nestjs/microservices';
 import { Test } from '@nestjs/testing';
 import { expect } from 'chai';
 import * as request from 'supertest';
 import { KafkaController } from '../src/kafka/kafka.controller';
+import { APP_FILTER } from '@nestjs/core';
+import { Observable, throwError } from 'rxjs';
+import { KafkaMessagesController } from '../src/kafka/kafka.messages.controller';
+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';
+
+@Catch()
+class KafkaExceptionFilter implements ExceptionFilter {
+  catch(exception: HttpException, host: ArgumentsHost): any {
+    console.log(exception);
+  }
+}
+@Catch()
+class RpcErrorFilter implements RpcExceptionFilter {
+  catch(exception: RpcException): Observable<any> {
+    console.log(exception);
+    return throwError(exception);
+  }
+}
 
 describe('Kafka transport', () => {
   let server;
   let app: INestApplication;
 
   it(`Start Kafka app`, async () => {
     const module = await Test.createTestingModule({
-      controllers: [KafkaController],
+      controllers: [
+        KafkaController,
+        KafkaMessagesController,
+      ],
+      providers: [
+        {
+          provide: APP_FILTER,
+          useClass: RpcErrorFilter,
+        },
+        {
+          provide: APP_FILTER,
+          useClass: KafkaExceptionFilter,
+        },
+      ],
     }).compile();
 
     app = module.createNestApplication();
@@ -37,30 +78,43 @@ describe('Kafka transport', () => {
       .expect(200, '15');
   }).timeout(50000);
 
-  // it(`/POST (async command sum)`, done => {
-  //   request(server)
-  //     .post('/?command=math.sum')
-  //     .send([1, 2, 3, 4, 5])
-  //     .end(() => {
-  //       setTimeout(() => {
-  //         expect(KafkaController.MATH_SUM).to.eq(15);
-  //         done();
-  //       }, 4000);
-  //     });
-  // }).timeout(5000);
-
   it(`/POST (async event notification)`, done => {
     request(server)
       .post('/notify')
-      .send([1, 2, 3, 4, 5])
+      .send()
       .end(() => {
-        setTimeout(() => {
-          expect(KafkaController.IS_NOTIFIED).to.be.true;
-          done();
-        }, 4000);
+        expect(KafkaController.IS_NOTIFIED).to.be.true;
+        done();
       });
   }).timeout(5000);
 
+  const userDto: UserDto = {
+    email: 'enriquebenavidesm@gmail.com',
+    name: 'Ben',
+    phone: '1112223331',
+    years: 33,
+  };
+  const newUser: UserEntity = new UserEntity(userDto);
+  const businessDto: BusinessDto = {
+    name: 'Example',
+    phone: '2233441122',
+    user: newUser,
+  };
+  it(`/POST (async command create user)`, () => {
+    return request(server)
+      .post('/user')
+      .send(userDto)
+      .expect(200);
+  }).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);
+
   after(`Stopping Kafka app`, async () => {
     await app.close();
   });

integration/microservices/src/kafka/dtos/business.dto.ts

@@ -0,0 +1,7 @@
+import { UserEntity } from '../entities/user.entity';
+
+export class BusinessDto {
+  name: string;
+  phone: string;
+  user: UserEntity;
+}

integration/microservices/src/kafka/dtos/user.dto.ts

@@ -0,0 +1,6 @@
+export class UserDto {
+  name: string;
+  email: string;
+  phone: string;
+  years: number;
+}

integration/microservices/src/kafka/entities/business.entity.ts

@@ -0,0 +1,19 @@
+import { UserEntity } from './user.entity';
+import { BusinessDto } from '../dtos/business.dto';
+
+export class BusinessEntity {
+  constructor(business: BusinessDto) {
+    this.id = Math.random() * 99999999;
+    this.name = business.name;
+    this.phone = business.phone;
+    this.createdBy = {
+      id: business.user.id,
+    };
+    this.created = new Date();
+  }
+  id: number;
+  name: string;
+  phone: string;
+  createdBy: Partial<UserEntity>;
+  created: Date;
+}

integration/microservices/src/kafka/entities/user.entity.ts

@@ -0,0 +1,18 @@
+import { UserDto } from '../dtos/user.dto';
+
+export class UserEntity {
+  constructor(user: UserDto) {
+    this.id = Math.random() * 99999999;
+    this.name = user.name;
+    this.email = user.email;
+    this.phone = user.phone;
+    this.years = user.years;
+    this.created = new Date();
+  }
+  id: number;
+  name: string;
+  email: string;
+  phone: string;
+  years: number;
+  created: Date;
+}

integration/microservices/src/kafka/kafka.controller.ts

@@ -1,20 +1,18 @@
-import { Body, Controller, HttpCode, Post, Query, OnModuleInit } from '@nestjs/common';
+import { Body, Controller, HttpCode, Post, Query } from '@nestjs/common';
 import {
   Client,
   ClientProxy,
-  EventPattern,
-  MessagePattern,
   Transport,
   MessageRequest,
 } from '@nestjs/microservices';
 import { Logger } from '@nestjs/common/services/logger.service';
-import * as util from 'util';
 
 import { Observable } from 'rxjs';
-import * as Bluebird from 'bluebird';
+import { UserDto } from './dtos/user.dto';
+import { BusinessDto } from './dtos/business.dto';
 
 @Controller()
-export class KafkaController implements OnModuleInit {
+export class KafkaController {
   protected readonly logger = new Logger(KafkaController.name);
   static IS_NOTIFIED = false;
   static MATH_SUM = 0;
@@ -29,10 +27,6 @@ export class KafkaController implements OnModuleInit {
   })
   private readonly client: ClientProxy;
 
-  public async onModuleInit(){
-    await this.client.connect();
-  }
-
   @Post()
   @HttpCode(200)
   @MessageRequest('math.sum', 'math.sum.reply')
@@ -46,28 +40,39 @@ export class KafkaController implements OnModuleInit {
         numbers: data,
       },
     }).toPromise();
-
-    // await Bluebird.delay(30000);
-
-    this.logger.error(util.format('@Query math.sum result %o', result));
-
     return result.value;
   }
 
-  @MessagePattern('math.sum')
-  mathSum(data: any){
-    this.logger.error(util.format('@MessagePattern math.sum data %o', data));
-
-    return (data.value.numbers || []).reduce((a, b) => a + b);
-  }
-
   @Post('notify')
   async sendNotification(): Promise<any> {
     return this.client.emit('notify', {notify: true});
   }
 
-  @EventPattern('notify')
-  eventHandler(data: any) {
-    KafkaController.IS_NOTIFIED = data.value.notify;
+  // Complex data to send.
+  @Post('/user')
+  @HttpCode(200)
+  @MessageRequest('user.create', 'user.create.reply')
+  async createUser(@Body() user: UserDto): Promise<Observable<any>> {
+    const result = await this.client.send('user.create', {
+      key: '1',
+      value: {
+        user,
+      },
+    }).toPromise();
+    return result.value;
+  }
+
+  // Complex data to send.
+  @Post('/business')
+  @HttpCode(200)
+  @MessageRequest('business.create', 'business.create.reply')
+  async createBusiness(@Body() business: BusinessDto) {
+    const result = await this.client.send('business.create', {
+      key: '1',
+      value: {
+        business,
+      },
+    }).toPromise();
+    return result.value;
   }
 }

integration/microservices/src/kafka/kafka.messages.controller.ts

@@ -0,0 +1,47 @@
+import { Controller } from '@nestjs/common';
+import { Client, ClientProxy, EventPattern, MessagePattern, MessageRequest, Transport } from '@nestjs/microservices';
+import { Logger } from '@nestjs/common/services/logger.service';
+import { KafkaController } from './kafka.controller';
+import { BusinessDto } from './dtos/business.dto';
+import { UserEntity } from './entities/user.entity';
+import { BusinessEntity } from './entities/business.entity';
+import * as util from 'util';
+import { UserDto } from './dtos/user.dto';
+
+@Controller()
+export class KafkaMessagesController {
+  protected readonly logger = new Logger(KafkaMessagesController.name);
+  static IS_NOTIFIED = false;
+  static MATH_SUM = 0;
+
+  @Client({
+    transport: Transport.KAFKA,
+    options: {
+      client: {
+        brokers: ['localhost:9092'],
+      },
+    },
+  })
+  private readonly client: ClientProxy;
+
+  @MessagePattern('math.sum')
+  mathSum(data: any){
+    return (data.value.numbers || []).reduce((a, b) => a + b);
+  }
+
+  @EventPattern('notify')
+  eventHandler(data: any) {
+    KafkaController.IS_NOTIFIED = data.value.notify;
+  }
+
+  // Complex data to send.
+  @MessagePattern('user.create')
+  async createUser(params: {value: { user: UserDto } }) {
+    return new UserEntity(params.value.user);
+  }
+
+  @MessagePattern('business.create')
+  async createBusiness(params: {value: {business: BusinessDto}}) {
+    return new BusinessEntity(params.value.business);
+  }
+}