Connect NestJS to a websocket server

Question

How can NestJS be use as a websocket client? I want to connect to a remote websocket server as a client using NestJS, but I didn't find any information about this implementation in the framework.

Answer 1

As Nestjs is simply a framework for Nodejs, so you need to find an NPM package that supports Websocket. For example, I use ws with @types/ws type definition, and create a Websocket client as a Nestjs service class:

// socket-client.ts
import { Injectable } from "@nestjs/common";
import * as WebSocket from "ws";

@Injectable()
export class WSService {
    // wss://echo.websocket.org is a test websocket server
    private ws = new WebSocket("wss://echo.websocket.org");

    constructor() {
        this.ws.on("open", () => {
            this.ws.send(Math.random())
        });

        this.ws.on("message", function(message) {
            console.log(message);
        });
    }

    send(data: any) {
        this.ws.send(data);
    }

    onMessage(handler: Function) {
        // ...
    }

    // ...
}

// app.module.ts
import { Module } from "@nestjs/common";
import { WSService } from "./socket-client";

@Module({
    providers: [WSService]
})
export class AppModule {}

Answer 2

I try it by another way. I write an adapter with socket.io-client. Then use this adapter in boostrap by method useWebSocketAdapter. After that i can write handle websocket event in gateway like the way working with socket server (use decorator @SubscribeMessage)

My Adapter file

import { WebSocketAdapter, INestApplicationContext } from '@nestjs/common';
import { MessageMappingProperties } from '@nestjs/websockets'

import * as SocketIoClient from 'socket.io-client';
import { isFunction, isNil } from '@nestjs/common/utils/shared.utils';
import { fromEvent, Observable } from 'rxjs';
import { filter, first, map, mergeMap, share, takeUntil } from 'rxjs/operators';

export class IoClientAdapter implements WebSocketAdapter {
    private io;
    constructor(private app: INestApplicationContext) {

    }

    create(port: number, options?: SocketIOClient.ConnectOpts) {
        const client = SocketIoClient("http://localhost:3000" , options || {})
        this.io = client;
        return client;
    }

    bindClientConnect(server: SocketIOClient.Socket, callback: Function) {
        this.io.on('connect', callback);
    }

    bindClientDisconnect(client: SocketIOClient.Socket, callback: Function) {
        console.log("it disconnect")
        //client.on('disconnect', callback);
    }

    public bindMessageHandlers(
        client: any,
        handlers: MessageMappingProperties[],
        transform: (data: any) => Observable<any>,
    ) {
        const disconnect$ = fromEvent(this.io, 'disconnect').pipe(
            share(),
            first(),
        );

        handlers.forEach(({ message, callback }) => {
            const source$ = fromEvent(this.io, message).pipe(
                mergeMap((payload: any) => {
                    const { data, ack } = this.mapPayload(payload);
                    return transform(callback(data, ack)).pipe(
                        filter((response: any) => !isNil(response)),
                        map((response: any) => [response, ack]),
                    );
                }),
                takeUntil(disconnect$),
            );
            source$.subscribe(([response, ack]) => {
                if (response.event) {
                    return client.emit(response.event, response.data);
                }
                isFunction(ack) && ack(response);
            });
        });
    }

    public mapPayload(payload: any): { data: any; ack?: Function } {
        if (!Array.isArray(payload)) {
            return { data: payload };
        }
        const lastElement = payload[payload.length - 1];
        const isAck = isFunction(lastElement);
        if (isAck) {
            const size = payload.length - 1;
            return {
                data: size === 1 ? payload[0] : payload.slice(0, size),
                ack: lastElement,
            };
        }
        return { data: payload };
    }

    close(server: SocketIOClient.Socket) {
        this.io.close()
    }
}

main.js

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import {IoClientAdapter} from './adapters/ioclient.adapter'

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.useWebSocketAdapter(new IoClientAdapter(app))
  await app.listen(3006);
  console.log(`Application is running on: ${await app.getUrl()}`);
}
bootstrap();

then Gateway

import {
  MessageBody,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
  WsResponse,
} from '@nestjs/websockets';
import { from, Observable } from 'rxjs';
import { map } from 'rxjs/operators';
import { Server } from 'socket.io';

@WebSocketGateway()
export class EventsGateway {
  @WebSocketServer()
  server: Server;

  @SubscribeMessage('hello')
  async identity(@MessageBody() data: number): Promise<number> {
      console.log(data)
    return data;
  }
}

It a trick, but look so cool. Message handler can write more like nestjs style.