nest/packages/microservices/server/server-nats.ts

import { Observable } from 'rxjs';
import {
  CONNECT_EVENT,
  ERROR_EVENT,
  NATS_DEFAULT_URL,
  NO_PATTERN_MESSAGE,
} from '../constants';
import { Client } from '../external/nats-client.interface';
import { CustomTransportStrategy, PacketId } from '../interfaces';
import {
  MicroserviceOptions,
  NatsOptions,
} from '../interfaces/microservice-configuration.interface';
import { ReadPacket } from '../interfaces/packet.interface';
import { Server } from './server';

let natsPackage: any = {};

export class ServerNats extends Server implements CustomTransportStrategy {
  private readonly url: string;
  private natsClient: Client;

  constructor(private readonly options: MicroserviceOptions['options']) {
    super();
    this.url =
      this.getOptionsProp<NatsOptions>(this.options, 'url') || NATS_DEFAULT_URL;

    natsPackage = this.loadPackage('nats', ServerNats.name);
  }

  public listen(callback: () => void) {
    this.natsClient = this.createNatsClient();
    this.handleError(this.natsClient);
    this.start(callback);
  }

  public start(callback?: () => void) {
    this.bindEvents(this.natsClient);
    this.natsClient.on(CONNECT_EVENT, callback);
  }

  public bindEvents(client: Client) {
    const registeredPatterns = [...this.messageHandlers.keys()];
    registeredPatterns.forEach(channel => {
      client.subscribe(
        channel,
        this.getMessageHandler(channel, client).bind(this),
      );
    });
  }

  public close() {
    this.natsClient && this.natsClient.close();
    this.natsClient = null;
  }

  public createNatsClient(): Client {
    const options = this.options || ({} as NatsOptions);
    return natsPackage.connect({
      ...(options as any),
      url: this.url,
      json: true,
    });
  }

  public getMessageHandler(channel: string, client: Client): Function {
    return async (buffer, replyTo: string) =>
      this.handleMessage(channel, buffer, client, replyTo);
  }

  public async handleMessage(
    channel: string,
    message: ReadPacket & PacketId,
    client: Client,
    replyTo: string,
  ) {
    const publish = this.getPublisher(client, replyTo, message.id);
    const handler = this.getHandlerByPattern(channel);
    if (!handler) {
      const status = 'error';
      return publish({ id: message.id, status, err: NO_PATTERN_MESSAGE });
    }
    const response$ = this.transformToObservable(
      await handler(message.data),
    ) as Observable<any>;
    response$ && this.send(response$, publish);
  }

  public getPublisher(publisher: Client, replyTo: string, id: string) {
    return response =>
      publisher.publish(replyTo, Object.assign(response, {
        id,
      }) as any);
  }

  public handleError(stream) {
    stream.on(ERROR_EVENT, err => this.logger.error(err));
  }
}