Merge pull request nestjs#974 from nestjs/bugfix/948-nats-wildcards

bugfix(@nestjs/microservices) fix nats wildcards

Author: kamilmysliwiec

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

@@ -1,8 +1,8 @@
-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 { Test } from '@nestjs/testing';
+import * as express from 'express';
+import * as request from 'supertest';
 import { NatsController } from '../src/nats/nats.controller';
 
 describe('NATS transport', () => {
@@ -19,7 +19,7 @@ describe('NATS transport', () => {
     app.connectMicroservice({
       transport: Transport.NATS,
       options: {
-        url: 'nats://localhost:4222'
+        url: 'nats://localhost:4222',
       },
     });
     await app.startAllMicroservicesAsync();
@@ -28,22 +28,22 @@ describe('NATS transport', () => {
 
   it(`/POST`, () => {
     return request(server)
-      .post('/?command=sum')
+      .post('/?command=math.sum')
       .send([1, 2, 3, 4, 5])
       .expect(200, '15');
   });
 
   it(`/POST (Promise/async)`, () => {
     return request(server)
-      .post('/?command=asyncSum')
+      .post('/?command=async.sum')
       .send([1, 2, 3, 4, 5])
       .expect(200)
       .expect(200, '15');
   });
 
   it(`/POST (Observable stream)`, () => {
     return request(server)
-      .post('/?command=streamSum')
+      .post('/?command=stream.sum')
       .send([1, 2, 3, 4, 5])
       .expect(200, '15');
   });

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

@@ -1,8 +1,8 @@
 import { Controller, Get } from '@nestjs/common';
 import {
   Client,
-  MessagePattern,
   ClientProxy,
+  MessagePattern,
   Transport,
 } from '@nestjs/microservices';
 import { Observable } from 'rxjs';
@@ -16,11 +16,11 @@ export class NatsBroadcastController {
   @Get('broadcast')
   multicats() {
     return this.client
-      .send<number>({ cmd: 'broadcast' }, {})
+      .send<number>('broadcast.test', {})
       .pipe(scan((a, b) => a + b), take(2));
   }
 
-  @MessagePattern({ cmd: 'broadcast' })
+  @MessagePattern('broadcast.*')
   replyBroadcast(): Observable<number> {
     return new Observable(observer => observer.next(1));
   }

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

@@ -1,11 +1,11 @@
-import { Controller, Get, Post, Body, Query, HttpCode } from '@nestjs/common';
+import { Body, Controller, HttpCode, Post, Query } from '@nestjs/common';
 import {
   Client,
-  MessagePattern,
   ClientProxy,
+  MessagePattern,
   Transport,
 } from '@nestjs/microservices';
-import { Observable, of, from } from 'rxjs';
+import { from, Observable, of } from 'rxjs';
 import { scan } from 'rxjs/operators';
 
 @Controller()
@@ -21,14 +21,14 @@ export class NatsController {
   @Post()
   @HttpCode(200)
   call(@Query('command') cmd, @Body() data: number[]): Observable<number> {
-    return this.client.send<number>({ cmd }, data);
+    return this.client.send<number>(cmd, data);
   }
 
   @Post('stream')
   @HttpCode(200)
   stream(@Body() data: number[]): Observable<number> {
     return this.client
-      .send<number>({ cmd: 'streaming' }, data)
+      .send<number>('streaming.sum', data)
       .pipe(scan((a, b) => a + b));
   }
 
@@ -38,7 +38,7 @@ export class NatsController {
     const send = async (tab: number[]) => {
       const expected = tab.reduce((a, b) => a + b);
       const result = await this.client
-        .send<number>({ cmd: 'sum' }, tab)
+        .send<number>('math.sum', tab)
         .toPromise();
 
       return result === expected;
@@ -48,22 +48,22 @@ export class NatsController {
       .reduce(async (a, b) => (await a) && (await b));
   }
 
-  @MessagePattern({ cmd: 'sum' })
+  @MessagePattern('math.*')
   sum(data: number[]): number {
     return (data || []).reduce((a, b) => a + b);
   }
 
-  @MessagePattern({ cmd: 'asyncSum' })
+  @MessagePattern('async.*')
   async asyncSum(data: number[]): Promise<number> {
     return (data || []).reduce((a, b) => a + b);
   }
 
-  @MessagePattern({ cmd: 'streamSum' })
+  @MessagePattern('stream.*')
   streamSum(data: number[]): Observable<number> {
     return of((data || []).reduce((a, b) => a + b));
   }
 
-  @MessagePattern({ cmd: 'streaming' })
+  @MessagePattern('streaming.*')
   streaming(data: number[]): Observable<number> {
     return from(data);
   }

packages/common/services/logger.service.ts

@@ -1,65 +1,93 @@
 import * as clc from 'cli-color';
+import { Injectable, Optional } from '../decorators';
 import { NestEnvironment } from '../enums/nest-environment.enum';
-import { Constructor } from '../utils/merge-with-values.util';
+import { isObject } from '../utils/shared.utils';
 
 declare const process;
 
 export interface LoggerService {
-  log(message: string);
-  error(message: string, trace: string);
-  warn(message: string);
+  log(message: any, context?: string);
+  error(message: any, trace?: string, context?: string);
+  warn(message: any, context?: string);
 }
 
+@Injectable()
 export class Logger implements LoggerService {
   private static prevTimestamp = null;
-  private static contextEnv = NestEnvironment.RUN;
+  private static contextEnvironment = NestEnvironment.RUN;
   private static logger: typeof Logger | LoggerService = Logger;
-
   private static readonly yellow = clc.xterm(3);
 
   constructor(
-    private readonly context: string,
-    private readonly isTimeDiffEnabled = false,
+    @Optional() private readonly context: string,
+    @Optional() private readonly isTimeDiffEnabled = false,
   ) {}
 
-  log(message: string) {
+  log(message: any, context?: string) {
     const { logger } = Logger;
+    if (logger === this) {
+      Logger.log(message, context || this.context, this.isTimeDiffEnabled);
+      return;
+    }
     logger &&
-      logger.log.call(logger, message, this.context, this.isTimeDiffEnabled);
+      logger.log.call(
+        logger,
+        message,
+        context || this.context,
+        this.isTimeDiffEnabled,
+      );
   }
 
-  error(message: string, trace = '') {
+  error(message: any, trace = '', context?: string) {
     const { logger } = Logger;
+    if (logger === this) {
+      Logger.error(
+        message,
+        trace,
+        context || this.context,
+        this.isTimeDiffEnabled,
+      );
+      return;
+    }
     logger &&
       logger.error.call(
         logger,
         message,
         trace,
-        this.context,
+        context || this.context,
         this.isTimeDiffEnabled,
       );
   }
 
-  warn(message: string) {
+  warn(message: any, context?: string) {
     const { logger } = Logger;
+    if (logger === this) {
+      Logger.warn(message, context || this.context, this.isTimeDiffEnabled);
+      return;
+    }
     logger &&
-      logger.warn.call(logger, message, this.context, this.isTimeDiffEnabled);
+      logger.warn.call(
+        logger,
+        message,
+        context || this.context,
+        this.isTimeDiffEnabled,
+      );
   }
 
   static overrideLogger(logger: LoggerService | boolean) {
     this.logger = logger ? (logger as LoggerService) : null;
   }
 
   static setMode(mode: NestEnvironment) {
-    this.contextEnv = mode;
+    this.contextEnvironment = mode;
   }
 
-  static log(message: string, context = '', isTimeDiffEnabled = true) {
+  static log(message: any, context = '', isTimeDiffEnabled = true) {
     this.printMessage(message, clc.green, context, isTimeDiffEnabled);
   }
 
   static error(
-    message: string,
+    message: any,
     trace = '',
     context = '',
     isTimeDiffEnabled = true,
@@ -68,22 +96,26 @@ export class Logger implements LoggerService {
     this.printStackTrace(trace);
   }
 
-  static warn(message: string, context = '', isTimeDiffEnabled = true) {
+  static warn(message: any, context = '', isTimeDiffEnabled = true) {
     this.printMessage(message, clc.yellow, context, isTimeDiffEnabled);
   }
 
   private static printMessage(
-    message: string,
-    color: (msg: string) => string,
+    message: any,
+    color: (message: string) => string,
     context: string = '',
     isTimeDiffEnabled?: boolean,
   ) {
-    if (Logger.contextEnv === NestEnvironment.TEST) return;
-
+    if (Logger.contextEnvironment === NestEnvironment.TEST) {
+      return void 0;
+    }
+    const output =
+      message && isObject(message) ? JSON.stringify(message, null, 2) : message;
     process.stdout.write(color(`[Nest] ${process.pid}   - `));
     process.stdout.write(`${new Date(Date.now()).toLocaleString()}   `);
-    process.stdout.write(this.yellow(`[${context}] `));
-    process.stdout.write(color(message));
+
+    context && process.stdout.write(this.yellow(`[${context}] `));
+    process.stdout.write(color(output));
 
     this.printTimestamp(isTimeDiffEnabled);
     process.stdout.write(`\n`);
@@ -100,8 +132,9 @@ export class Logger implements LoggerService {
   }
 
   private static printStackTrace(trace: string) {
-    if (this.contextEnv === NestEnvironment.TEST || !trace) return;
-
+    if (this.contextEnvironment === NestEnvironment.TEST || !trace) {
+      return void 0;
+    }
     process.stdout.write(trace);
     process.stdout.write(`\n`);
   }

packages/microservices/client/client-mqtt.ts

@@ -85,7 +85,7 @@ export class ClientMqtt extends ClientProxy {
   ): Function {
     try {
       const packet = this.assignPacketId(partialPacket);
-      const pattern = JSON.stringify(partialPacket.pattern);
+      const pattern = this.normalizePattern(partialPacket.pattern);
       const responseChannel = this.getResPatternName(pattern);
       const responseCallback = this.createResponseCallback(packet, callback);
 

packages/microservices/client/client-nats.ts

@@ -21,14 +21,6 @@ export class ClientNats extends ClientProxy {
     natsPackage = loadPackage('nats', ClientNats.name);
   }
 
-  public getAckPatternName(pattern: string): string {
-    return `${pattern}_ack`;
-  }
-
-  public getResPatternName(pattern: string): string {
-    return `${pattern}_res`;
-  }
-
   public close() {
     this.natsClient && this.natsClient.close();
     this.natsClient = null;
@@ -81,21 +73,24 @@ export class ClientNats extends ClientProxy {
       });
     };
   }
+
   protected publish(
     partialPacket: ReadPacket,
     callback: (packet: WritePacket) => any,
   ): Function {
     try {
       const packet = this.assignPacketId(partialPacket);
-      const pattern = JSON.stringify(partialPacket.pattern);
-      const responseChannel = this.getResPatternName(pattern);
+      const channel = this.normalizePattern(partialPacket.pattern);
 
-      const subscriptionHandler = this.createSubscriptionHandler(packet, callback);
-      const subscriptionId = this.natsClient.subscribe(
-        responseChannel,
+      const subscriptionHandler = this.createSubscriptionHandler(
+        packet,
+        callback,
+      );
+      const subscriptionId = this.natsClient.request(
+        channel,
+        packet as any,
         subscriptionHandler,
       );
-      this.natsClient.publish(this.getAckPatternName(pattern), packet as any);
       return () => this.natsClient.unsubscribe(subscriptionId);
     } catch (err) {
       callback({ err });

packages/microservices/client/client-proxy.ts

@@ -1,9 +1,21 @@
-import { isNil } from '@nestjs/common/utils/shared.utils';
-import { defer, fromEvent, merge, Observable, Observer, throwError as _throw } from 'rxjs';
+import { isNil, isString } from '@nestjs/common/utils/shared.utils';
+import {
+  defer,
+  fromEvent,
+  merge,
+  Observable,
+  Observer,
+  throwError as _throw,
+} from 'rxjs';
 import { map, mergeMap, take } from 'rxjs/operators';
 import { CONNECT_EVENT, ERROR_EVENT } from '../constants';
 import { InvalidMessageException } from '../exceptions/errors/invalid-message.exception';
-import { ClientOptions, PacketId, ReadPacket, WritePacket } from '../interfaces';
+import {
+  ClientOptions,
+  PacketId,
+  ReadPacket,
+  WritePacket,
+} from '../interfaces';
 
 export abstract class ClientProxy {
   public abstract connect(): Promise<any>;
@@ -74,4 +86,8 @@ export abstract class ClientProxy {
   ) {
     return obj && obj.options ? obj.options[prop as any] : defaultValue;
   }
+
+  protected normalizePattern<T = any>(pattern: T): string {
+    return pattern && isString(pattern) ? pattern : JSON.stringify(pattern);
+  }
 }