feat(microservices): support nats v2

Author: kamilmysliwiec

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

@@ -19,13 +19,13 @@ describe('NATS transport', () => {
     app.connectMicroservice({
       transport: Transport.NATS,
       options: {
-        url: 'nats://0.0.0.0:4222',
+        servers: 'nats://0.0.0.0:4222',
       },
     });
     app.connectMicroservice({
       transport: Transport.NATS,
       options: {
-        url: 'nats://0.0.0.0:4222',
+        servers: 'servers://0.0.0.0:4222',
       },
     });
     await app.startAllMicroservices();

integration/microservices/e2e/disconnected-client.spec.ts

@@ -46,7 +46,7 @@ describe('Disconnected client', () => {
       .send({
         transport: Transport.NATS,
         options: {
-          url: 'nats://localhost:4224',
+          servers: 'nats://localhost:4224',
         },
       })
       .expect(408);

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

@@ -22,7 +22,7 @@ describe('NATS transport', () => {
     app.connectMicroservice({
       transport: Transport.NATS,
       options: {
-        url: 'nats://0.0.0.0:4222',
+        servers: 'nats://0.0.0.0:4222',
       },
     });
     await app.startAllMicroservices();

integration/microservices/src/disconnected.controller.ts

@@ -18,13 +18,15 @@ export class DisconnectedClientController {
       .send<number, number[]>({ cmd: 'none' }, [1, 2, 3])
       .pipe(
         /*tap(
-        console.log.bind(console, 'data'),
-        console.error.bind(console, 'error'),
-      ),*/
+          console.log.bind(console, 'data'),
+          console.error.bind(console, 'error'),
+        ),*/
         catchError(error => {
           const { code } = error || { code: 'CONN_ERR' };
           return throwError(
-            code === 'ECONNREFUSED' || code === 'CONN_ERR'
+            code === 'ECONNREFUSED' ||
+              code === 'CONN_ERR' ||
+              code === 'CONNECTION_REFUSED'
               ? new RequestTimeoutException('ECONNREFUSED')
               : new InternalServerErrorException(),
           );

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

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

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

@@ -24,7 +24,7 @@ export class NatsController {
   client: ClientProxy = ClientProxyFactory.create({
     transport: Transport.NATS,
     options: {
-      url: 'nats://localhost:4222',
+      servers: 'nats://localhost:4222',
     },
   });
 

package-lock.json

@@ -149,7 +149,7 @@
     "mqtt": "4.2.6",
     "multer": "1.4.2",
     "mysql": "2.18.1",
-    "nats": "1.4.12",
+    "nats": "2.0.0",
     "nodemon": "2.0.7",
     "nyc": "15.1.0",
     "point-of-view": "4.13.0",

package.json

@@ -1,68 +1,75 @@
 import { Logger } from '@nestjs/common/services/logger.service';
 import { loadPackage } from '@nestjs/common/utils/load-package.util';
-import { share } from 'rxjs/operators';
-import { CONN_ERR, ERROR_EVENT, NATS_DEFAULT_URL } from '../constants';
-import { Client } from '../external/nats-client.interface';
+import { NATS_DEFAULT_URL } from '../constants';
+import { NatsResponseJSONDeserializer } from '../deserializers/nats-response-json.deserializer';
+import { Client, NatsMsg } from '../external/nats-client.interface';
 import { NatsOptions, PacketId, ReadPacket, WritePacket } from '../interfaces';
+import { NatsJSONSerializer } from '../serializers/nats-json.serializer';
 import { ClientProxy } from './client-proxy';
 
-let natsPackage: any = {};
+let natsPackage = {} as any;
 
 export class ClientNats extends ClientProxy {
   protected readonly logger = new Logger(ClientProxy.name);
-  protected readonly url: string;
   protected natsClient: Client;
-  protected connection: Promise<any>;
 
   constructor(protected readonly options: NatsOptions['options']) {
     super();
-    this.url = this.getOptionsProp(this.options, 'url') || NATS_DEFAULT_URL;
     natsPackage = loadPackage('nats', ClientNats.name, () => require('nats'));
 
     this.initializeSerializer(options);
     this.initializeDeserializer(options);
   }
 
-  public close() {
-    this.natsClient && this.natsClient.close();
+  public async close() {
+    await this.natsClient?.close();
     this.natsClient = null;
-    this.connection = null;
   }
 
   public async connect(): Promise<any> {
     if (this.natsClient) {
-      return this.connection;
+      return this.natsClient;
     }
-    this.natsClient = this.createClient();
-    this.handleError(this.natsClient);
-
-    this.connection = await this.connect$(this.natsClient)
-      .pipe(share())
-      .toPromise();
-    return this.connection;
+    this.natsClient = await this.createClient();
+    this.handleStatusUpdates(this.natsClient);
+    return this.natsClient;
   }
 
-  public createClient(): Client {
+  public createClient(): Promise<Client> {
     const options: any = this.options || ({} as NatsOptions);
     return natsPackage.connect({
+      servers: NATS_DEFAULT_URL,
       ...options,
-      url: this.url,
-      json: true,
     });
   }
 
-  public handleError(client: Client) {
-    client.addListener(
-      ERROR_EVENT,
-      (err: any) => err.code !== CONN_ERR && this.logger.error(err),
-    );
+  public async handleStatusUpdates(client: Client) {
+    for await (const status of client.status()) {
+      const data =
+        status.data && typeof status.data === 'object'
+          ? JSON.stringify(status.data)
+          : status.data;
+      if (status.type === 'disconnect' || status.type === 'error') {
+        this.logger.error(
+          `NatsError: type: "${status.type}", data: "${data}".`,
+        );
+      } else {
+        this.logger.log(`NatsStatus: type: "${status.type}", data: "${data}".`);
+      }
+    }
   }
 
   public createSubscriptionHandler(
     packet: ReadPacket & PacketId,
     callback: (packet: WritePacket) => any,
-  ): Function {
-    return (rawPacket: unknown) => {
+  ) {
+    return (error: unknown | undefined, natsMsg: NatsMsg) => {
+      if (error) {
+        return callback({
+          err: error,
+        });
+      }
+      const rawPacket = natsMsg.data;
       const message = this.deserializer.deserialize(rawPacket);
       if (message.id && message.id !== packet.id) {
         return undefined;
@@ -95,12 +102,14 @@ export class ClientNats extends ClientProxy {
         packet,
         callback,
       );
-      const subscriptionId = this.natsClient.request(
-        channel,
-        serializedPacket as any,
-        subscriptionHandler,
-      );
-      return () => this.natsClient.unsubscribe(subscriptionId);
+      this.natsClient.publish(channel, serializedPacket, {
+        reply: packet.id,
+      });
+      const subscription = this.natsClient.subscribe(packet.id, {
+        callback: subscriptionHandler,
+      });
+
+      return () => subscription.unsubscribe();
     } catch (err) {
       callback({ err });
     }
@@ -110,10 +119,22 @@ export class ClientNats extends ClientProxy {
     const pattern = this.normalizePattern(packet.pattern);
     const serializedPacket = this.serializer.serialize(packet);
 
-    return new Promise<void>((resolve, reject) =>
-      this.natsClient.publish(pattern, serializedPacket as any, err =>
-        err ? reject(err) : resolve(),
-      ),
-    );
+    return new Promise<void>((resolve, reject) => {
+      try {
+        this.natsClient.publish(pattern, serializedPacket);
+        resolve();
+      } catch (err) {
+        reject(err);
+      }
+    });
+  }
+
+  protected initializeSerializer(options: NatsOptions['options']) {
+    this.serializer = options?.serializer ?? new NatsJSONSerializer();
+  }
+
+  protected initializeDeserializer(options: NatsOptions['options']) {
+    this.deserializer =
+      options?.deserializer ?? new NatsResponseJSONDeserializer();
   }
 }

packages/microservices/client/client-nats.ts

@@ -1,6 +1,6 @@
 import { BaseRpcContext } from './base-rpc.context';
 
-type NatsContextArgs = [string];
+type NatsContextArgs = [string, any];
 
 export class NatsContext extends BaseRpcContext<NatsContextArgs> {
   constructor(args: NatsContextArgs) {
@@ -13,4 +13,11 @@ export class NatsContext extends BaseRpcContext<NatsContextArgs> {
   getSubject() {
     return this.args[0];
   }
+
+  /**
+   * Returns message headers (if exist).
+   */
+  getHeaders() {
+    return this.args[1];
+  }
 }