将 NestJS 连接到 websocket 服务器

Posted 2023-03-10

【中文标题】将 NestJS 连接到 websocket 服务器

【英文标题】：Connect NestJS to a websocket server

【发布时间】：2020-08-22 10:07:48

【问题描述】：

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

【参考方案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 

【讨论】：

这里不能使用内置的 websocket 功能吗？ docs.nestjs.com/websockets/gateways

【参考方案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风格。