feat(microservices): add status, unwrap, on, and other features

integration/microservices/e2e/sum-rpc-async.spec.ts

@@ -4,18 +4,18 @@ import {
   Injectable,
   Module,
 } from '@nestjs/common';
+import { NestFactory } from '@nestjs/core';
 import {
   AsyncOptions,
   ClientTCP,
   ClientsModule,
   MessagePattern,
   MicroserviceOptions,
   Payload,
-  TcpClientOptions,
+  TcpOptions,
   Transport,
 } from '@nestjs/microservices';
 import { expect } from 'chai';
-import { NestFactory } from '@nestjs/core';
 
 let port: number;
 
@@ -25,7 +25,7 @@ do {
 
 @Injectable()
 class RpcOptionsProvider {
-  getOptions(): TcpClientOptions {
+  getOptions(): TcpOptions {
     return {
       transport: Transport.TCP,
       options: {

integration/microservices/src/tcp-tls/app.module.ts

@@ -1,13 +1,13 @@
-import { Module, Injectable } from '@nestjs/common';
-import { AppController } from './app.controller';
+import { Injectable, Module } from '@nestjs/common';
 import {
+  ClientOptions,
   ClientsModule,
-  Transport,
   ClientsModuleOptionsFactory,
-  ClientOptions,
   ClientTCP,
   RpcException,
+  Transport,
 } from '@nestjs/microservices';
+import { AppController } from './app.controller';
 
 import * as fs from 'fs';
 import * as path from 'path';

packages/common/interfaces/nest-microservice.interface.ts

@@ -1,3 +1,4 @@
+import { Observable } from 'rxjs';
 import { ExceptionFilter } from './exceptions/exception-filter.interface';
 import { CanActivate } from './features/can-activate.interface';
 import { NestInterceptor } from './features/nest-interceptor.interface';
@@ -19,46 +20,73 @@ export interface INestMicroservice extends INestApplicationContext {
   listen(): Promise<any>;
 
   /**
-   * Register Ws Adapter which will be used inside Gateways.
-   * Use when you want to override default `socket.io` library.
+   * Registers a web socket adapter that will be used for Gateways.
+   * Use to override the default `socket.io` library.
    *
    * @param {WebSocketAdapter} adapter
    * @returns {this}
    */
   useWebSocketAdapter(adapter: WebSocketAdapter): this;
 
   /**
-   * Registers exception filters as global filters (will be used within every message pattern handler)
+   * Registers global exception filters (will be used for every pattern handler).
    *
    * @param {...ExceptionFilter} filters
    */
   useGlobalFilters(...filters: ExceptionFilter[]): this;
 
   /**
-   * Registers pipes as global pipes (will be used within every message pattern handler)
+   * Registers global pipes (will be used for every pattern handler).
    *
    * @param {...PipeTransform} pipes
    */
   useGlobalPipes(...pipes: PipeTransform<any>[]): this;
 
   /**
-   * Registers interceptors as global interceptors (will be used within every message pattern handler)
+   * Registers global interceptors (will be used for every pattern handler).
    *
    * @param {...NestInterceptor} interceptors
    */
   useGlobalInterceptors(...interceptors: NestInterceptor[]): this;
 
   /**
-   * Registers guards as global guards (will be used within every message pattern handler)
+   * Registers global guards (will be used for every pattern handler).
    *
    * @param {...CanActivate} guards
    */
   useGlobalGuards(...guards: CanActivate[]): this;
 
   /**
-   * Terminates the application
+   * Terminates the application.
    *
    * @returns {Promise<void>}
    */
   close(): Promise<void>;
+
+  /**
+   * Returns an observable that emits status changes.
+   *
+   * @returns {Observable<string>}
+   */
+  status: Observable<string>;
+
+  /**
+   * Registers an event listener for the given event.
+   * @param event Event name
+   * @param callback Callback to be executed when the event is emitted
+   */
+  on<
+    EventsMap extends Record<string, Function> = Record<string, Function>,
+    EventKey extends keyof EventsMap = keyof EventsMap,
+    EventCallback extends EventsMap[EventKey] = EventsMap[EventKey],
+  >(
+    event: EventKey,
+    callback: EventCallback,
+  ): void;
+
+  /**
+   * Returns an instance of the underlying server/broker instance,
+   * or a group of servers if there are more than one.
+   */
+  unwrap<T>(): T;
 }

packages/core/scanner.ts

@@ -389,7 +389,11 @@ export class DependenciesScanner {
     // Skip "InternalCoreModule" from calculating distance
     modulesGenerator.next();
 
-    const calculateDistance = (moduleRef: Module, distance = 1, modulesStack = []) => {
+    const calculateDistance = (
+      moduleRef: Module,
+      distance = 1,
+      modulesStack = [],
+    ) => {
       const localModulesStack = [...modulesStack];
       if (!moduleRef || localModulesStack.includes(moduleRef)) {
         return;

packages/microservices/client/client-grpc.ts

@@ -6,23 +6,42 @@ import { GRPC_DEFAULT_PROTO_LOADER, GRPC_DEFAULT_URL } from '../constants';
 import { InvalidGrpcPackageException } from '../errors/invalid-grpc-package.exception';
 import { InvalidGrpcServiceException } from '../errors/invalid-grpc-service.exception';
 import { InvalidProtoDefinitionException } from '../errors/invalid-proto-definition.exception';
-import { ClientGrpc, GrpcOptions } from '../interfaces';
-import { ClientProxy } from './client-proxy';
-import { GRPC_CANCELLED } from './constants';
 import { ChannelOptions } from '../external/grpc-options.interface';
 import { getGrpcPackageDefinition } from '../helpers';
+import { ClientGrpc, GrpcOptions } from '../interfaces';
+import { ClientProxy } from './client-proxy';
 
-let grpcPackage: any = {};
-let grpcProtoLoaderPackage: any = {};
+const GRPC_CANCELLED = 'Cancelled';
+
+// To enable type safety for gRPC. This cant be uncommented by default
+// because it would require the user to install the @grpc/grpc-js package even if they dont use gRPC
+// Otherwise, TypeScript would fail to compile the code.
+//
+// type GrpcClient = import('@grpc/grpc-js').Client;
+// let grpcPackage = {} as typeof import('@grpc/grpc-js');
+// let grpcProtoLoaderPackage = {} as typeof import('@grpc/proto-loader');
+
+type GrpcClient = any;
+let grpcPackage = {} as any;
+let grpcProtoLoaderPackage = {} as any;
 
 /**
  * @publicApi
  */
-export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
+export class ClientGrpcProxy
+  extends ClientProxy<never, never>
+  implements ClientGrpc
+{
   protected readonly logger = new Logger(ClientProxy.name);
   protected readonly clients = new Map<string, any>();
   protected readonly url: string;
-  protected grpcClients = [];
+  protected grpcClients: GrpcClient[] = [];
+
+  get status(): never {
+    throw new Error(
+      'The "status" attribute is not supported by the gRPC transport',
+    );
+  }
 
   constructor(protected readonly options: GrpcOptions['options']) {
     super();
@@ -367,4 +386,15 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
       'Method is not supported in gRPC mode. Use ClientGrpc instead (learn more in the documentation).',
     );
   }
+
+  public on<EventKey extends never = never, EventCallback = any>(
+    event: EventKey,
+    callback: EventCallback,
+  ) {
+    throw new Error('Method is not supported in gRPC mode.');
+  }
+
+  public unwrap<T>(): T {
+    throw new Error('Method is not supported in gRPC mode.');
+  }
 }