test(@nestjs) update unit tests

package.json

@@ -55,7 +55,7 @@
     "reflect-metadata": "^0.1.10",
     "rxjs": "^5.5.6",
     "rxjs-grpc": "^0.1.6",
-    "socket.io": "^2.0.3",
+    "socket.io": "^2.0.4",
     "trouter": "^1.0.0"
   },
   "devDependencies": {
@@ -69,7 +69,7 @@
     "@types/redis": "^0.12.36",
     "@types/reflect-metadata": "0.0.5",
     "@types/sinon": "^1.16.36",
-    "@types/socket.io": "^1.4.29",
+    "@types/socket.io": "^1.4.32",
     "awesome-typescript-loader": "^3.0.0-beta.18",
     "body-parser": "^1.17.2",
     "chai": "^3.5.0",

src/common/interfaces/websockets/web-socket-adapter.interface.ts

@@ -1,17 +1,15 @@
 import { Observable } from 'rxjs/Observable';
 
 export interface WebSocketAdapter {
-  create(port: number);
-  createWithNamespace?(port: number, namespace: string, server?);
+  create(port: number, options?: any & { namespace?: string; server?: any });
   bindClientConnect(server, callback: (...args) => void);
   bindClientDisconnect?(client, callback: (...args) => void);
   bindMessageHandlers(
     client,
     handler: {
-      message: string;
-      callback: (...args) => Observable<any> | Promise<any> | void;
+      message: any;
+      callback: (...args) => Observable<any> | Promise<any> | any;
     }[],
     process: (data) => Observable<any>,
   );
-  bindMiddleware?(server, middleware: (socket, next) => void);
 }

src/common/test/interceptors/files.interceptor.spec.ts

@@ -29,5 +29,17 @@ describe('FilesInterceptor', () => {
       expect(arraySpy.called).to.be.true;
       expect(arraySpy.calledWith(fieldName, maxCount)).to.be.true;
     });
+    it('should transform exception', async () => {
+      const fieldName = 'file';
+      const target = new (FilesInterceptor(fieldName));
+      const err = {};
+      const callback = (req, res, next) => next(err);
+
+      (target as any).upload = {
+        array: () => callback,
+      };
+      const req = {};
+      expect(target.intercept(req, null, stream$)).to.eventually.throw();
+    });
   });
 });

src/core/errors/exceptions/circular-dependency.exception.ts

@@ -0,0 +1,9 @@
+import { RuntimeException } from './runtime.exception';
+
+export class CircularDependencyException extends RuntimeException {
+  constructor(context: string) {
+    super(
+      `A circular dependency has been detected inside ${context}. Please, make sure that each side of a bidirectional relationships are decorated with "forwardRef()".`,
+    );
+  }
+}

src/core/scanner.ts

@@ -13,8 +13,9 @@ import { Type } from '@nestjs/common/interfaces/type.interface';
 import { MetadataScanner } from '../core/metadata-scanner';
 import { DynamicModule } from '@nestjs/common';
 import { ApplicationConfig } from './application-config';
-import { isNil } from '@nestjs/common/utils/shared.utils';
+import { isNil, isUndefined } from '@nestjs/common/utils/shared.utils';
 import { APP_INTERCEPTOR, APP_PIPE, APP_GUARD, APP_FILTER } from './constants';
+import { CircularDependencyException } from './errors/exceptions/circular-dependency.exception';
 
 interface ApplicationProviderWrapper {
   moduleToken: string;
@@ -58,14 +59,14 @@ export class DependenciesScanner {
     const modules = this.container.getModules();
 
     modules.forEach(({ metatype }, token) => {
-      this.reflectRelatedModules(metatype, token);
+      this.reflectRelatedModules(metatype, token, metatype.name);
       this.reflectComponents(metatype, token);
       this.reflectControllers(metatype, token);
       this.reflectExports(metatype, token);
     });
   }
 
-  public reflectRelatedModules(module: Type<any>, token: string) {
+  public reflectRelatedModules(module: Type<any>, token: string, context: string) {
     const modules = [
       ...this.reflectMetadata(module, metadata.MODULES),
       ...this.container.getDynamicMetadataByToken(
@@ -77,7 +78,7 @@ export class DependenciesScanner {
         metadata.IMPORTS as 'imports',
       ),
     ];
-    modules.map(related => this.storeRelatedModule(related, token));
+    modules.map(related => this.storeRelatedModule(related, token, context));
   }
 
   public reflectComponents(module: Type<any>, token: string) {
@@ -189,8 +190,11 @@ export class DependenciesScanner {
     return descriptor ? Reflect.getMetadata(key, descriptor.value) : undefined;
   }
 
-  public storeRelatedModule(related: any, token: string) {
-    if (related.forwardRef) {
+  public storeRelatedModule(related: any, token: string, context: string) {
+    if (isUndefined(related)) {
+      throw new CircularDependencyException(context);
+    }
+    if (related && related.forwardRef) {
       return this.container.addRelatedModule(related.forwardRef(), token);
     }
     this.container.addRelatedModule(related, token);

src/core/test/scanner.spec.ts

@@ -156,7 +156,7 @@ describe('DependenciesScanner', () => {
       const module = { forwardRef: sinon.stub().returns({}) };
 
       sinon.stub(container, 'addRelatedModule').returns({});
-      scanner.storeRelatedModule(module as any, [] as any);
+      scanner.storeRelatedModule(module as any, [] as any, 'test');
       expect(module.forwardRef.called).to.be.true;
     });
   });

src/microservices/client/client-grpc.ts

@@ -8,6 +8,7 @@ import { ClientGrpc } from './../interfaces';
 import { Observable } from 'rxjs/Observable';
 import { InvalidGrpcServiceException } from '../exceptions/invalid-grpc-service.exception';
 import { InvalidGrpcPackageException } from '../exceptions/invalid-grpc-package.exception';
+import { InvalidProtoDefinitionException } from '../exceptions/invalid-proto-definition.exception';
 
 export interface GrpcService {
   [name: string]: (...args: any[]) => Observable<any>;
@@ -84,9 +85,7 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
   }
 
   public createClient(): any {
-    const grpcContext = grpc.load(
-      this.getOptionsProp<GrpcOptions>(this.options, 'protoPath'),
-    );
+    const grpcContext = this.loadProto();
     const packageName = this.getOptionsProp<GrpcOptions>(
       this.options,
       'package',
@@ -98,6 +97,18 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
     return grpcPackage;
   }
 
+  public loadProto(): grpc.GrpcObject {
+    try {
+      const context = grpc.load(
+        this.getOptionsProp<GrpcOptions>(this.options, 'protoPath'),
+      );
+      return context;
+    }
+    catch (e) {
+      throw new InvalidProtoDefinitionException();
+    }
+  }
+
   public lookupPackage(root: any, packageName: string) {
     /** Reference: https://github.com/kondi/rxjs-grpc */
     let pkg = root;

src/microservices/client/client-mqtt.ts

@@ -23,7 +23,7 @@ export class ClientMqtt extends ClientProxy {
       this.getOptionsProp<MqttOptions>(this.options, 'url') || MQTT_DEFAULT_URL;
   }
 
-  protected async publish(
+  protected publish(
     partialPacket: ReadPacket,
     callback: (packet: WritePacket) => any,
   ) {

src/microservices/client/client-nats.ts

@@ -3,8 +3,12 @@ import { ClientProxy } from './client-proxy';
 import { Logger } from '@nestjs/common/services/logger.service';
 import { ClientOptions } from '../interfaces/client-metadata.interface';
 import { NATS_DEFAULT_URL, ERROR_EVENT, CONNECT_EVENT } from './../constants';
-import { WritePacket, NatsOptions } from './../interfaces';
-import { ReadPacket, PacketId } from 'src/microservices';
+import {
+  WritePacket,
+  NatsOptions,
+  ReadPacket,
+  PacketId,
+} from './../interfaces';
 
 export class ClientNats extends ClientProxy {
   private readonly logger = new Logger(ClientProxy.name);
@@ -28,28 +32,30 @@ export class ClientNats extends ClientProxy {
     const pattern = JSON.stringify(partialPacket.pattern);
     const responseChannel = this.getResPatternName(pattern);
 
-    const subscriptionId = this.natsClient.subscribe(
-      responseChannel,
-      (message: WritePacket & PacketId) => {
-        if (message.id !== packet.id) {
-          return void 0;
-        }
-        const { err, response, isDisposed } = message;
-        if (isDisposed || err) {
-          callback({
-            err,
-            response: null,
-            isDisposed: true,
-          });
-          return this.natsClient.unsubscribe(subscriptionId);
-        }
+    const subscriptionHandler = (message: WritePacket & PacketId) => {
+      if (message.id !== packet.id) {
+        return void 0;
+      }
+      const { err, response, isDisposed } = message;
+      if (isDisposed || err) {
         callback({
           err,
-          response,
+          response: null,
+          isDisposed: true,
         });
-      },
+        return this.natsClient.unsubscribe(subscriptionId);
+      }
+      callback({
+        err,
+        response,
+      });
+    };
+    const subscriptionId = this.natsClient.subscribe(
+      responseChannel,
+      subscriptionHandler,
     );
     this.natsClient.publish(this.getAckPatternName(pattern), packet as any);
+    return subscriptionHandler;
   }
 
   public getAckPatternName(pattern: string): string {

src/microservices/client/client-redis.ts

@@ -25,7 +25,8 @@ export class ClientRedis extends ClientProxy {
 
   constructor(private readonly options: ClientOptions) {
     super();
-    this.url = this.getOptionsProp<RedisOptions>(options, 'url') || REDIS_DEFAULT_URL;
+    this.url =
+      this.getOptionsProp<RedisOptions>(options, 'url') || REDIS_DEFAULT_URL;
   }
 
   protected async publish(
@@ -62,12 +63,16 @@ export class ClientRedis extends ClientProxy {
     };
     this.subClient.on(MESSAGE_EVENT, responseCallback);
     this.subClient.subscribe(responseChannel);
-    await new Promise(resolve =>
-      this.subClient.on(
-        SUBSCRIBE,
-        channel => channel === responseChannel && resolve(),
-      ),
-    );
+    await new Promise(resolve => {
+      const handler = channel => {
+        if (channel && channel !== responseChannel) {
+          return void 0;
+        }
+        this.subClient.removeListener(SUBSCRIBE, handler);
+        resolve();
+      };
+      this.subClient.on(SUBSCRIBE, handler);
+    });
 
     this.pubClient.publish(
       this.getAckPatternName(pattern),

src/microservices/client/client-tcp.ts

@@ -14,8 +14,7 @@ import {
   ERROR_EVENT,
   CLOSE_EVENT,
 } from './../constants';
-import { WritePacket } from './../interfaces';
-import { ReadPacket, PacketId } from 'src/microservices';
+import { WritePacket, ReadPacket, PacketId } from './../interfaces';
 
 export class ClientTCP extends ClientProxy {
   private readonly logger = new Logger(ClientTCP.name);

src/microservices/context/rpc-context-creator.ts

@@ -29,7 +29,7 @@ export class RpcContextCreator {
     instance: Controller,
     callback: (data) => Observable<any>,
     module,
-  ): (data) => Promise<Observable<any>> {
+  ): (...args) => Promise<Observable<any>> {
     const exceptionHandler = this.exceptionFiltersContext.create(
       instance,
       callback,

src/microservices/exceptions/invalid-proto-definition.exception.ts

@@ -0,0 +1,7 @@
+import { RuntimeException } from '@nestjs/core/errors/exceptions/runtime.exception';
+
+export class InvalidProtoDefinitionException extends RuntimeException {
+  constructor() {
+    super('Invalid .proto definition (file not found?)');
+  }
+}

src/microservices/server/server-grpc.ts

@@ -26,12 +26,12 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
   }
 
   public async start(callback?: () => void) {
-    await this.bindEvents(this.grpcClient);
+    await this.bindEvents();
     this.grpcClient.start();
     callback();
   }
 
-  public async bindEvents(client: any) {
+  public async bindEvents() {
     const grpcContext = grpc.load(
       this.getOptionsProp<GrpcOptions>(this.options, 'protoPath'),
     );

src/microservices/server/server-tcp.ts

@@ -21,10 +21,10 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
   private isExplicitlyTerminated = false;
   private retryAttemptsCount = 0;
 
-  constructor(private readonly config: MicroserviceOptions) {
+  constructor(private readonly options: MicroserviceOptions) {
     super();
     this.port =
-      this.getOptionsProp<TcpOptions>(config, 'port') ||
+      this.getOptionsProp<TcpOptions>(options, 'port') ||
       TCP_DEFAULT_PORT;
     this.init();
   }
@@ -71,12 +71,12 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
     if (
       this.isExplicitlyTerminated ||
       !this.getOptionsProp<TcpOptions>(
-        this.config,
+        this.options,
         'retryAttempts',
       ) ||
       this.retryAttemptsCount >=
         this.getOptionsProp<TcpOptions>(
-          this.config,
+          this.options,
           'retryAttempts',
         )
     ) {
@@ -86,7 +86,7 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
     return setTimeout(
       () => this.server.listen(this.port),
       this.getOptionsProp<TcpOptions>(
-        this.config,
+        this.options,
         'retryDelay',
       ) || 0,
     );