将 NestJS 连接到 websocket 服务器

Question

如何将 NestJS 用作 websocket 客户端？我想使用 NestJS 作为客户端连接到远程 websocket 服务器，但我没有在框架中找到有关此实现的任何信息。

Answer 1

由于 Nestjs 只是 Nodejs 的一个框架，所以需要找一个支持 Websocket 的 NPM 包。例如，我使用ws 和@types/ws 类型定义，并创建一个Websocket 客户端作为Nestjs 服务类：

// 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

我用另一种方式尝试。我用socket.io-client 编写了一个适配器。然后通过方法useWebSocketAdapter 在boostrap 中使用这个适配器。之后我可以在网关中编写处理 websocket 事件，就像使用套接字服务器的方式一样（使用装饰器 @SubscribeMessage）

我的适配器文件

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();

然后是网关

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;
  }
}

这是一个技巧，但看起来很酷。消息处理程序可以写得更像nestjs风格。