refactor() resolve conflicts

Author: kamilmysliwiec

.gitignore

@@ -5,6 +5,7 @@ node_modules/
 /.idea
 /.awcache
 /.vscode
+*.code-workspace
 
 # bundle
 packages/**/*.d.ts

integration/docker-compose.yml

@@ -31,15 +31,42 @@ services:
       - "3306:3306"
     restart: always
   mongodb:
+    container_name: test-mongodb
     image: mongo:latest
     environment:
       - MONGODB_DATABASE="test"
     ports:
       - 27017:27017
   rabbit:
+    container_name: test-rabbit
     hostname: rabbit
     image: "rabbitmq:management"
     ports:
       - "15672:15672"
       - "5672:5672"
     tty: true
+  zookeeper:
+    container_name: test-zookeeper
+    hostname: zookeeper
+    image: confluentinc/cp-zookeeper:5.3.0
+    ports:
+      - "2181:2181"
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+  kafka:
+    container_name: test-kafka
+    hostname: kafka
+    image: confluentinc/cp-kafka:5.3.0
+    depends_on:
+      - zookeeper
+    ports:
+      - "29092:29092"
+      - "9092:9092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

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

@@ -0,0 +1,140 @@
+import { INestApplication } from '@nestjs/common';
+import { Transport } from '@nestjs/microservices';
+import { Test } from '@nestjs/testing';
+import { expect } from 'chai';
+import * as request from 'supertest';
+import { BusinessDto } from '../src/kafka/dtos/business.dto';
+import { UserDto } from '../src/kafka/dtos/user.dto';
+import { UserEntity } from '../src/kafka/entities/user.entity';
+import { KafkaController } from '../src/kafka/kafka.controller';
+import { KafkaMessagesController } from '../src/kafka/kafka.messages.controller';
+
+describe('Kafka transport', () => {
+  let server;
+  let app: INestApplication;
+
+  it(`Start Kafka app`, async () => {
+    const module = await Test.createTestingModule({
+      controllers: [KafkaController, KafkaMessagesController],
+    }).compile();
+
+    app = module.createNestApplication();
+    server = app.getHttpAdapter().getInstance();
+
+    app.connectMicroservice({
+      transport: Transport.KAFKA,
+      options: {
+        client: {
+          brokers: ['localhost:9092'],
+        },
+      },
+    });
+    await app.startAllMicroservicesAsync();
+    await app.init();
+  }).timeout(30000);
+
+  it(`/POST (sync sum kafka message)`, () => {
+    return request(server)
+      .post('/mathSumSyncKafkaMessage')
+      .send([1, 2, 3, 4, 5])
+      .expect(200)
+      .expect(200, '15');
+  });
+
+  it(`/POST (sync sum kafka(ish) message without key and only the value)`, () => {
+    return request(server)
+      .post('/mathSumSyncWithoutKey')
+      .send([1, 2, 3, 4, 5])
+      .expect(200)
+      .expect(200, '15');
+  });
+
+  it(`/POST (sync sum plain object)`, () => {
+    return request(server)
+      .post('/mathSumSyncPlainObject')
+      .send([1, 2, 3, 4, 5])
+      .expect(200)
+      .expect(200, '15');
+  });
+
+  it(`/POST (sync sum array)`, () => {
+    return request(server)
+      .post('/mathSumSyncArray')
+      .send([1, 2, 3, 4, 5])
+      .expect(200)
+      .expect(200, '15');
+  });
+
+  it(`/POST (sync sum string)`, () => {
+    return request(server)
+      .post('/mathSumSyncString')
+      .send([1, 2, 3, 4, 5])
+      .expect(200)
+      .expect(200, '15');
+  });
+
+  it(`/POST (sync sum number)`, () => {
+    return request(server)
+      .post('/mathSumSyncNumber')
+      .send([12345])
+      .expect(200)
+      .expect(200, '15');
+  });
+
+  it(`/POST (async event notification)`, done => {
+    request(server)
+      .post('/notify')
+      .send()
+      .end(() => {
+        setTimeout(() => {
+          expect(KafkaController.IS_NOTIFIED).to.be.true;
+          done();
+        }, 1000);
+      });
+  });
+
+  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 (sync command create user)`, () => {
+    return request(server)
+      .post('/user')
+      .send(userDto)
+      .expect(200);
+  });
+
+  it(`/POST (sync command create business`, () => {
+    return request(server)
+      .post('/business')
+      .send(businessDto)
+      .expect(200);
+  });
+
+  it(`/POST (sync command create user) Concurrency Test`, async () => {
+    const promises = [];
+    for (let concurrencyKey = 0; concurrencyKey < 100; concurrencyKey++) {
+      const innerUserDto = JSON.parse(JSON.stringify(userDto));
+      innerUserDto.name += `+${concurrencyKey}`;
+      promises.push(
+        request(server)
+          .post('/user')
+          .send(userDto)
+          .expect(200),
+      );
+    }
+    await Promise.all(promises);
+  });
+
+  after(`Stopping Kafka app`, async () => {
+    await app.close();
+  });
+}).timeout(30000);

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

@@ -0,0 +1,152 @@
+import { Body, Controller, HttpCode, OnModuleInit, Post } from '@nestjs/common';
+import { Logger } from '@nestjs/common/services/logger.service';
+import { Client, ClientKafka, Transport } from '@nestjs/microservices';
+import { Observable } from 'rxjs';
+import { BusinessDto } from './dtos/business.dto';
+import { UserDto } from './dtos/user.dto';
+
+@Controller()
+export class KafkaController implements OnModuleInit {
+  protected readonly logger = new Logger(KafkaController.name);
+  static IS_NOTIFIED = false;
+  static MATH_SUM = 0;
+
+  @Client({
+    transport: Transport.KAFKA,
+    options: {
+      client: {
+        brokers: ['localhost:9092'],
+      },
+    },
+  })
+  private readonly client: ClientKafka;
+
+  onModuleInit() {
+    const requestPatterns = [
+      'math.sum.sync.kafka.message',
+      'math.sum.sync.without.key',
+      'math.sum.sync.plain.object',
+      'math.sum.sync.array',
+      'math.sum.sync.string',
+      'math.sum.sync.number',
+      'user.create',
+      'business.create',
+    ];
+
+    requestPatterns.forEach(pattern => {
+      this.client.subscribeToResponseOf(pattern);
+    });
+  }
+
+  // sync send kafka message
+  @Post('mathSumSyncKafkaMessage')
+  @HttpCode(200)
+  async mathSumSyncKafkaMessage(
+    @Body() data: number[],
+  ): Promise<Observable<any>> {
+    const result = await this.client
+      .send('math.sum.sync.kafka.message', {
+        key: '1',
+        value: {
+          numbers: data,
+        },
+      })
+      .toPromise();
+    return result;
+  }
+
+  // sync send kafka(ish) message without key and only the value
+  @Post('mathSumSyncWithoutKey')
+  @HttpCode(200)
+  async mathSumSyncWithoutKey(
+    @Body() data: number[],
+  ): Promise<Observable<any>> {
+    const result = await this.client
+      .send('math.sum.sync.without.key', {
+        value: {
+          numbers: data,
+        },
+      })
+      .toPromise();
+    return result;
+  }
+
+  // sync send message without key or value
+  @Post('mathSumSyncPlainObject')
+  @HttpCode(200)
+  async mathSumSyncPlainObject(
+    @Body() data: number[],
+  ): Promise<Observable<any>> {
+    const result = await this.client
+      .send('math.sum.sync.plain.object', {
+        numbers: data,
+      })
+      .toPromise();
+    return result;
+  }
+
+  // sync send message without key or value
+  @Post('mathSumSyncArray')
+  @HttpCode(200)
+  async mathSumSyncArray(@Body() data: number[]): Promise<Observable<any>> {
+    const result = await this.client
+      .send('math.sum.sync.array', data)
+      .toPromise();
+    return result;
+  }
+
+  @Post('mathSumSyncString')
+  @HttpCode(200)
+  async mathSumSyncString(@Body() data: number[]): Promise<Observable<any>> {
+    // this.logger.error(util.format('mathSumSyncString() data: %o', data));
+    const result = await this.client
+      .send('math.sum.sync.string', data.toString())
+      .toPromise();
+    return result;
+  }
+
+  @Post('mathSumSyncNumber')
+  @HttpCode(200)
+  async mathSumSyncNumber(@Body() data: number[]): Promise<Observable<any>> {
+    const result = await this.client
+      .send('math.sum.sync.number', data[0])
+      .toPromise();
+    return result;
+  }
+
+  // async notify
+  @Post('notify')
+  async sendNotification(): Promise<any> {
+    return this.client.emit('notify', { notify: true });
+  }
+
+  // Complex data to send.
+  @Post('/user')
+  @HttpCode(200)
+  async createUser(@Body() user: UserDto): Promise<Observable<any>> {
+    const result = await this.client
+      .send('user.create', {
+        key: '1',
+        value: {
+          user,
+        },
+      })
+      .toPromise();
+    return result;
+  }
+
+  // Complex data to send.
+  @Post('/business')
+  @HttpCode(200)
+  async createBusiness(@Body() business: BusinessDto) {
+    const result = await this.client
+      .send('business.create', {
+        key: '1',
+        value: {
+          business,
+        },
+      })
+      .toPromise();
+    return result;
+  }
+}