feature(@nestjs/microservices) add grpc, mqtt, microservices refactor

integration/docker-compose.yml

@@ -2,17 +2,23 @@ version: "3"
 
 services:
   redis:
-    container_name: redis_integration
+    container_name: test-redis
     image: redis
     ports:
       - "6379:6379"
     restart: always
   nats:
-    container_name: nats_integration
+    container_name: test-nats
     image: nats
-    expose:
-      - "4222"
     ports:
       - "8222:8222"
       - "4222:4222"
+      - "6222:6222"
+    restart: always
+  mqtt:
+    container_name: test-mqtt
+    image: toke/mosquitto
+    ports:
+      - "1883:1883"
+      - "9001:9001"
     restart: always

integration/microservices/e2e/multicast-mqtt.spec.ts

@@ -0,0 +1,38 @@
+import * as express from 'express';
+import * as request from 'supertest';
+import { Test } from '@nestjs/testing';
+import { INestApplication } from '@nestjs/common';
+import { Transport } from '@nestjs/microservices';
+import { MqttMulticastController } from '../src/mqtt/mqtt-multicast.controller';
+
+describe('MQTT transport', () => {
+  let server;
+  let app: INestApplication;
+
+  beforeEach(async () => {
+    const module = await Test.createTestingModule({
+      controllers: [MqttMulticastController],
+    }).compile();
+
+    server = express();
+    app = module.createNestApplication(server);
+    app.connectMicroservice({
+      transport: Transport.MQTT,
+    });
+    app.connectMicroservice({
+      transport: Transport.MQTT,
+    });
+    await app.startAllMicroservicesAsync();
+    await app.init();
+  });
+
+  it(`Multicast (2 subscribers)`, () => {
+    return request(server)
+      .get('/multicast')
+      .expect(200, '2');
+  });
+
+  afterEach(async () => {
+    await app.close();
+  });
+});

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

@@ -0,0 +1,41 @@
+import * as express from 'express';
+import * as request from 'supertest';
+import { Test } from '@nestjs/testing';
+import { INestApplication } from '@nestjs/common';
+import { Transport } from '@nestjs/microservices';
+import { join } from 'path';
+import { GrpcController } from '../src/grpc/grpc.controller';
+
+describe('GRPC transport', () => {
+  let server;
+  let app: INestApplication;
+
+  beforeEach(async () => {
+    const module = await Test.createTestingModule({
+      controllers: [GrpcController],
+    }).compile();
+
+    server = express();
+    app = module.createNestApplication(server);
+    app.connectMicroservice({
+      transport: Transport.GRPC,
+      options: {
+        package: 'math',
+        protoPath: join(__dirname, './../src/grpc/math.proto')
+      },
+    });
+    await app.startAllMicroservicesAsync();
+    await app.init();
+  });
+
+  it(`/POST`, () => {
+    return request(server)
+      .post('/')
+      .send([1, 2, 3, 4, 5])
+      .expect(200, { result: '15' });
+  });
+
+  afterEach(async () => {
+    await app.close();
+  });
+});

integration/microservices/e2e/sum-stan.spec.ts → integration/microservices/e2e/sum-mqtt.spec.ts

@@ -1,23 +1,23 @@
-/*import * as express from 'express';
+import * as express from 'express';
 import * as request from 'supertest';
 import { Test } from '@nestjs/testing';
 import { INestApplication } from '@nestjs/common';
 import { Transport } from '@nestjs/microservices';
-import { NatsController } from '../src/nats/nats.controller';
+import { MqttController } from '../src/mqtt/mqtt.controller';
 
-describe('STAN transport', () => {
+describe('MQTT transport', () => {
   let server;
   let app: INestApplication;
 
   beforeEach(async () => {
     const module = await Test.createTestingModule({
-      controllers: [NatsController],
+      controllers: [MqttController],
     }).compile();
 
     server = express();
     app = module.createNestApplication(server);
     app.connectMicroservice({
-      transport: Transport.STAN,
+      transport: Transport.MQTT,
     });
     await app.startAllMicroservicesAsync();
     await app.init();
@@ -45,6 +45,19 @@ describe('STAN transport', () => {
       .expect(200, '15');
   });
 
+  it(`/POST (concurrent)`, () => {
+    return request(server)
+      .post('/concurrent')
+      .send([
+        [1, 2, 3, 4, 5],
+        [6, 7, 8, 9, 10],
+        [11, 12, 13, 14, 15],
+        [16, 17, 18, 19, 20],
+        [21, 22, 23, 24, 25],
+      ])
+      .expect(200, 'true');
+  });
+
   it(`/POST (streaming)`, () => {
     return request(server)
       .post('/stream')
@@ -56,4 +69,3 @@ describe('STAN transport', () => {
     await app.close();
   });
 });
-*/

integration/microservices/src/grpc/grpc.controller.ts

@@ -0,0 +1,38 @@
+import { Controller, Post, Body, HttpCode } from '@nestjs/common';
+import {
+  Client,
+  MessagePattern,
+  ClientProxy,
+  Transport,
+  GrpcRoute,
+  ClientGrpc,
+} from '@nestjs/microservices';
+import { Observable } from 'rxjs/Observable';
+import { of } from 'rxjs/observable/of';
+import { join } from 'path';
+
+@Controller()
+export class GrpcController {
+  @Client({
+    transport: Transport.GRPC,
+    options: {
+      package: 'math',
+      protoPath: join(__dirname, 'math.proto'),
+    }
+  })
+  client: ClientGrpc;
+
+  @Post()
+  @HttpCode(200)
+  call(@Body() data: number[]): Observable<number> {
+    const svc = this.client.getService<any>('Math');
+    return svc.Sum(data);
+  }
+
+  @GrpcRoute('Math', 'Sum')
+  async sum({ data }: { data: number[] }): Promise<any> {
+    return of({
+      result: data.reduce((a, b) => a + b),
+    });
+  }
+}

integration/microservices/src/grpc/math.proto

@@ -0,0 +1,15 @@
+syntax = "proto3";
+
+package math;
+
+service Math {
+  rpc Sum (RequestSum) returns (SumResult) {}
+}
+
+message SumResult {
+  required int32 result = 1;
+}
+
+message RequestSum {
+  repeated int32 data = 1;
+}

integration/microservices/src/mqtt/mqtt-multicast.controller.ts

@@ -0,0 +1,27 @@
+import { Controller, Get } from '@nestjs/common';
+import {
+  Client,
+  MessagePattern,
+  ClientProxy,
+  Transport,
+} from '@nestjs/microservices';
+import { Observable } from 'rxjs/Observable';
+import { scan, take } from 'rxjs/operators';
+
+@Controller()
+export class MqttMulticastController {
+  @Client({ transport: Transport.MQTT })
+  client: ClientProxy;
+
+  @Get('multicast')
+  multicats() {
+    return this.client
+      .send<number>({ cmd: 'multicast' }, {})
+      .pipe(scan((a, b) => a + b), take(2));
+  }
+
+  @MessagePattern({ cmd: 'multicast' })
+  replyMulticast(): Observable<number> {
+    return new Observable(observer => observer.next(1));
+  }
+}

integration/microservices/src/stan/stan.controller.ts → integration/microservices/src/mqtt/mqtt.controller.ts

@@ -11,8 +11,8 @@ import { from } from 'rxjs/observable/from';
 import { scan } from 'rxjs/operators';
 
 @Controller()
-export class NatsController {
-  @Client({ transport: Transport.STAN })
+export class MqttController {
+  @Client({ transport: Transport.MQTT })
   client: ClientProxy;
 
   @Post()

integration/microservices/src/nats/nats-multicast.controller.ts

@@ -10,7 +10,7 @@ import { scan, take } from 'rxjs/operators';
 
 @Controller()
 export class NatsMulticastController {
-  @Client({ transport: Transport.NATS, url: 'nats://localhost:4222' })
+  @Client({ transport: Transport.NATS })
   client: ClientProxy;
 
   @Get('multicast')

integration/microservices/src/nats/nats.controller.ts

@@ -12,7 +12,7 @@ import { scan } from 'rxjs/operators';
 
 @Controller()
 export class NatsController {
-  @Client({ transport: Transport.NATS, url: 'nats://localhost:4222' })
+  @Client({ transport: Transport.NATS })
   client: ClientProxy;
 
   @Post()

lib/common/constants.d.ts

@@ -1,25 +1,25 @@
 export declare const metadata: {
-    MODULES: string;
-    IMPORTS: string;
-    COMPONENTS: string;
-    PROVIDERS: string;
-    CONTROLLERS: string;
-    EXPORTS: string;
+  MODULES: string;
+  IMPORTS: string;
+  COMPONENTS: string;
+  PROVIDERS: string;
+  CONTROLLERS: string;
+  EXPORTS: string;
 };
-export declare const SHARED_MODULE_METADATA = "__sharedModule__";
-export declare const GLOBAL_MODULE_METADATA = "__globalModule__";
-export declare const PATH_METADATA = "path";
-export declare const PARAMTYPES_METADATA = "design:paramtypes";
-export declare const SELF_DECLARED_DEPS_METADATA = "self:paramtypes";
-export declare const METHOD_METADATA = "method";
-export declare const ROUTE_ARGS_METADATA = "__routeArguments__";
-export declare const CUSTOM_ROUTE_AGRS_METADATA = "__customRouteArgs__";
-export declare const EXCEPTION_FILTERS_METADATA = "__exceptionFilters__";
-export declare const FILTER_CATCH_EXCEPTIONS = "__filterCatchExceptions__";
-export declare const PIPES_METADATA = "__pipes__";
-export declare const GUARDS_METADATA = "__guards__";
-export declare const RENDER_METADATA = "__renderTemplate__";
-export declare const INTERCEPTORS_METADATA = "__interceptors__";
-export declare const HTTP_CODE_METADATA = "__httpCode__";
-export declare const GATEWAY_MIDDLEWARES = "__gatewayMiddlewares";
-export declare const MODULE_PATH = "__module_path__";
+export declare const SHARED_MODULE_METADATA = '__sharedModule__';
+export declare const GLOBAL_MODULE_METADATA = '__globalModule__';
+export declare const PATH_METADATA = 'path';
+export declare const PARAMTYPES_METADATA = 'design:paramtypes';
+export declare const SELF_DECLARED_DEPS_METADATA = 'self:paramtypes';
+export declare const METHOD_METADATA = 'method';
+export declare const ROUTE_ARGS_METADATA = '__routeArguments__';
+export declare const CUSTOM_ROUTE_AGRS_METADATA = '__customRouteArgs__';
+export declare const EXCEPTION_FILTERS_METADATA = '__exceptionFilters__';
+export declare const FILTER_CATCH_EXCEPTIONS = '__filterCatchExceptions__';
+export declare const PIPES_METADATA = '__pipes__';
+export declare const GUARDS_METADATA = '__guards__';
+export declare const RENDER_METADATA = '__renderTemplate__';
+export declare const INTERCEPTORS_METADATA = '__interceptors__';
+export declare const HTTP_CODE_METADATA = '__httpCode__';
+export declare const GATEWAY_MIDDLEWARES = '__gatewayMiddlewares';
+export declare const MODULE_PATH = '__module_path__';

lib/common/index.d.ts

@@ -1,6 +1,29 @@
 export * from './decorators';
 export * from './enums';
-export { NestModule, INestApplication, INestMicroservice, NestMiddleware, FunctionMiddleware, MiddlewaresConsumer, OnModuleInit, ExceptionFilter, WebSocketAdapter, PipeTransform, Paramtype, ArgumentMetadata, OnModuleDestroy, ExecutionContext, CanActivate, RpcExceptionFilter, WsExceptionFilter, NestInterceptor, DynamicModule, INestApplicationContext, HttpServer, HttpServerFactory } from './interfaces';
+export {
+  NestModule,
+  INestApplication,
+  INestMicroservice,
+  NestMiddleware,
+  FunctionMiddleware,
+  MiddlewaresConsumer,
+  OnModuleInit,
+  ExceptionFilter,
+  WebSocketAdapter,
+  PipeTransform,
+  Paramtype,
+  ArgumentMetadata,
+  OnModuleDestroy,
+  ExecutionContext,
+  CanActivate,
+  RpcExceptionFilter,
+  WsExceptionFilter,
+  NestInterceptor,
+  DynamicModule,
+  INestApplicationContext,
+  HttpServer,
+  HttpServerFactory,
+} from './interfaces';
 export * from './interceptors';
 export * from './services/logger.service';
 export * from './pipes';

lib/common/interfaces/nest-application.interface.d.ts

@@ -18,40 +18,12 @@ export interface INestApplication extends INestApplicationContext {
      * @returns void
      */
     use(...args: any[]): this;
-    /**
-     * A wrapper function around native `express.set()` method.
-     * Example `app.set('trust proxy', 'loopback')`
-     *
-     * @returns void
-     */
-    set(...args: any[]): this;
-    /**
-     * A wrapper function around native `express.engine()` method.
-     * Example `app.engine('mustache', mustacheExpress())`
-     *
-     * @returns void
-     */
-    engine(...args: any[]): this;
-    /**
-     * A wrapper function around native `express.enable()` method.
-     * Example `app.enable('x-powered-by')`
-     *
-     * @returns void
-     */
-    enable(...args: any[]): this;
     /**
      * Enables CORS (Cross-Origin Resource Sharing)
      *
      * @returns void
      */
     enableCors(options?: CorsOptions): this;
-    /**
-     * A wrapper function around native `express.disable()` method.
-     * Example `app.disable('x-powered-by')`
-     *
-     * @returns void
-     */
-    disable(...args: any[]): this;
     /**
      * Starts the application.
      *