update

libs/utils/src/utils.module.ts

@@ -5,6 +5,4 @@ import { UtilsService } from './utils.service';
   providers: [UtilsService],
   exports: [UtilsService],
 })
-export class UtilsModule {
-
-}
+export class UtilsModule {}

package.json

@@ -27,6 +27,8 @@
     "@nestjs/platform-socket.io": "^8.4.4",
     "@nestjs/platform-ws": "^8.4.4",
     "@nestjs/websockets": "^8.4.4",
+    "buffer": "^6.0.3",
+    "node-datachannel": "^0.3.1",
     "reflect-metadata": "^0.1.13",
     "rimraf": "^3.0.2",
     "rxjs": "^7.2.0"

src/main.ts

@@ -1,5 +1,6 @@
 import { NestFactory } from '@nestjs/core';
 import { AppModule } from './app.module';
+// import { WsAdapter } from '@nestjs/platform-ws';
 import { WsAdapter } from './ws-adapter';
 import { Logger } from '@nestjs/common';
 

src/meta.gateway.ts

@@ -6,10 +6,17 @@ import {
   OnGatewayConnection,
   OnGatewayDisconnect,
 } from '@nestjs/websockets';
+
 import { Server } from 'ws';
 import * as WebSocket from 'ws';
 
+import { PeerConnection, initLogger, DataChannel } from 'node-datachannel';
+import { Buffer } from 'buffer';
 import { Logger } from '@nestjs/common';
+import * as path from 'path';
+import { createReadStream } from 'fs';
+
+initLogger('Debug');
 
 @WebSocketGateway(3100, {
   transports: ['websocket'],
@@ -20,20 +27,194 @@ import { Logger } from '@nestjs/common';
 export class MetaGateway
   implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
   private logger: Logger = new Logger('MetaGateway');
-  // @WebSocketServer() server;
+  private peer: PeerConnection = null;
+  private timer: NodeJS.Timeout;
+  private _webrtcInterval: NodeJS.Timeout;
+  private heartBeatFlag: number;
+  private gameChanel: DataChannel;
+  @WebSocketServer() server: Server;
+
   // @SubscribeMessage('message')
-  // handleMessage(client: any, payload: any): string {
-  //   console.log('payload', payload);
-  //   return 'Hello world!';
+  // handleMessage(client: any, payload: any) {
+  //   this.logger.log(`payload: ${JSON.stringify(payload)}`);
   // }
+
+  @SubscribeMessage('init')
+  handleInit(client: any, payload: any) {
+    this.logger.log(`init: ${JSON.stringify(payload)}`);
+  }
+
+  @SubscribeMessage('heartbeat')
+  handleHeartBeat(client: any, payload: any) {
+    this.logger.log(`heartbeat: ${JSON.stringify(payload)}`);
+    // console.log('hb', payload);
+    this.heartBeatFlag = payload;
+    const pong = {
+      channel_id: '',
+      client_os: '',
+      data: payload,
+      fe_version: '',
+      id: 'heartbeat',
+      packet_id: '',
+      room_id: '',
+      session_id: '',
+      trace_id: '',
+      user_id: '',
+    };
+    if (this.gameChanel.isOpen) {
+      const heartPack = new DataView(new ArrayBuffer(4));
+      heartPack.setUint32(0, 2009889916);
+      this.gameChanel.sendMessageBinary(Buffer.from(heartPack.buffer));
+    }
+    return pong;
+  }
+
+  @SubscribeMessage('init_webrtc')
+  handleInitWebRtc(client: any, payload: any): void {
+    console.log('handleInitWebRtc');
+
+    this.peer = new PeerConnection('roomTest', {
+      iceServers: ['stun:120.24.252.95:3478'],
+    });
+
+    this.peer.onLocalDescription((sdp, type) => {
+      console.warn('peer SDP:', sdp, ' Type:', type);
+      const offer = { sdp, type };
+      const offerFormat = {
+        id: 'offer',
+        data: Buffer.from(JSON.stringify(offer)).toString('base64'),
+      };
+      console.log('send', offerFormat);
+      client.send(JSON.stringify(offerFormat));
+      // return '';
+    });
+
+    this.peer.onLocalCandidate((candidate, mid) => {
+      console.warn('peer Candidate:', candidate);
+      const iceRes = {
+        candidate,
+        sdpMid: mid,
+        sdpMLineIndex: 0,
+      };
+
+      const res = {
+        channel_id: '',
+        client_os: '',
+        data: Buffer.from(JSON.stringify(iceRes)).toString('base64'),
+        fe_version: '',
+        id: 'ice_candidate',
+        packet_id: '',
+        room_id: '',
+        session_id: '',
+        trace_id: '',
+        user_id: '',
+      };
+
+      // client.send(JSON.stringify(res));
+    });
+    this.peer.onStateChange((state) => {
+      console.log('peer-State:', state);
+    });
+    this.peer.onGatheringStateChange((state) => {
+      console.log('GatheringState:', state);
+    });
+    this.gameChanel = this.peer.createDataChannel('start-game');
+
+    this.peer.onDataChannel((dc) => {
+      console.log('onDataChannel', dc);
+    });
+
+    this.gameChanel.onOpen(() => {
+      console.log('channel is open');
+      clearInterval(this.timer);
+      let i = 0;
+      const paths = path.join(__dirname, '../ws/video/earth');
+      console.error('__dirname', __dirname);
+      console.error('paths', paths);
+      if (this.gameChanel.isOpen()) {
+        console.log('gameChanel', this.gameChanel.isOpen());
+        this.sendWertcHeartPack(this.gameChanel);
+      }
+      this.timer = setInterval(() => {
+        if (i < 10) {
+          const steam = createReadStream(paths + `/${i}`);
+          steam.on('data', (data: Buffer) => {
+            this.gameChanel.sendMessageBinary(data);
+          });
+        }
+        i++;
+      }, 2000);
+    });
+    this.gameChanel.onClosed(() => {
+      console.log('gameChanel close');
+      this.stopSendWertcHeartPack();
+    });
+    this.gameChanel.onError(() => {
+      console.log('gameChanel close');
+      this.stopSendWertcHeartPack();
+    });
+  }
+
+  sendWertcHeartPack(channel: DataChannel) {
+    const heartPack = new DataView(new ArrayBuffer(4));
+    heartPack.setUint32(0, 2009889916);
+    this._webrtcInterval = setInterval(() => {
+      channel.sendMessageBinary(Buffer.from(heartPack.buffer));
+    }, 100);
+  }
+
+  stopSendWertcHeartPack(): void {
+    clearInterval(this._webrtcInterval);
+  }
+
+  @SubscribeMessage('ice_candidate')
+  handlerIceCandidate(client: any, payload: any) {
+    const iceCandidate = Buffer.from(payload, 'base64').toString('utf-8');
+    const candidate = JSON.parse(iceCandidate);
+    console.error('收到ice_candidate', candidate);
+    this.peer.addRemoteCandidate(candidate.candidate, candidate.sdpMid);
+  }
+
+  @SubscribeMessage('answer')
+  handerAnswer(client: any, payload: any) {
+    const answer = Buffer.from(payload, 'base64').toString('utf-8');
+    console.log('answer', answer);
+    const clientAnswer = JSON.parse(answer);
+    this.peer.setLocalDescription(clientAnswer.sdp);
+    this.peer.setRemoteDescription(clientAnswer.sdp, clientAnswer.type);
+  }
+
+  @SubscribeMessage('start')
+  handlerWebrtcStart(client: any, payload: any) {
+    console.log('start', payload)
+
+  }
+
   afterInit(server: Server) {
     this.logger.log('Init');
   }
 
   handleConnection(client: WebSocket, ...args: any[]) {
     this.logger.log(`Client connected: ${client.id}`);
+    const connected = {
+      channel_id: '',
+      client_os: '',
+      data: '',
+      fe_version: '',
+      id: 'init',
+      packet_id: '',
+      room_id: '',
+      session_id: '',
+      trace_id: '',
+      user_id: '',
+    };
+    const tt = JSON.stringify(connected);
+    // console.log('tt', tt);
+
+    client.send(tt);
   }
   handleDisconnect(client: WebSocket) {
     this.logger.log(`Client disconnected: ${client.id}`);
+    this.peer && this.peer.close();
   }
 }

src/ws-adapter.ts

@@ -1,50 +1,204 @@
-import * as WebSocket from 'ws';
-import { WebSocketAdapter, INestApplicationContext } from '@nestjs/common';
-import { MessageMappingProperties } from '@nestjs/websockets';
-import { Observable, fromEvent, EMPTY } from 'rxjs';
-import { mergeMap, filter } from 'rxjs/operators';
+import { INestApplicationContext, Logger } from '@nestjs/common';
+import { loadPackage } from '@nestjs/common/utils/load-package.util';
+import { AbstractWsAdapter } from '@nestjs/websockets';
+import {
+  CLOSE_EVENT,
+  CONNECTION_EVENT,
+  ERROR_EVENT,
+} from '@nestjs/websockets/constants';
+import { MessageMappingProperties } from '@nestjs/websockets/gateway-metadata-explorer';
+import * as http from 'http';
+import { EMPTY, fromEvent, Observable } from 'rxjs';
+import { filter, first, mergeMap, share, takeUntil } from 'rxjs/operators';
 
-export class WsAdapter implements WebSocketAdapter {
-  constructor(private app: INestApplicationContext) { }
+let wsPackage: any = {};
 
-  create(port: number, options: any = {}): any {
-    return new WebSocket.Server({ port, ...options });
+enum READY_STATE {
+  CONNECTING_STATE = 0,
+  OPEN_STATE = 1,
+  CLOSING_STATE = 2,
+  CLOSED_STATE = 3,
+}
+
+type HttpServerRegistryKey = number;
+type HttpServerRegistryEntry = any;
+type WsServerRegistryKey = number;
+type WsServerRegistryEntry = any[];
+
+const UNDERLYING_HTTP_SERVER_PORT = 0;
+
+export class WsAdapter extends AbstractWsAdapter {
+  protected readonly logger = new Logger(WsAdapter.name);
+  protected readonly httpServersRegistry = new Map<
+    HttpServerRegistryKey,
+    HttpServerRegistryEntry
+  >();
+  protected readonly wsServersRegistry = new Map<
+    WsServerRegistryKey,
+    WsServerRegistryEntry
+  >();
+
+  constructor(appOrHttpServer?: INestApplicationContext | any) {
+    super(appOrHttpServer);
+    wsPackage = loadPackage('ws', 'WsAdapter', () => require('ws'));
   }
 
-  bindClientConnect(server, callback: Function) {
-    server.on('connection', callback);
+  public create(
+    port: number,
+    options?: Record<string, any> & { namespace?: string; server?: any },
+  ) {
+    const { server, ...wsOptions } = options;
+    if (wsOptions?.namespace) {
+      const error = new Error(
+        '"WsAdapter" does not support namespaces. If you need namespaces in your project, consider using the "@nestjs/platform-socket.io" package instead.',
+      );
+      this.logger.error(error);
+      throw error;
+    }
+
+    if (port === UNDERLYING_HTTP_SERVER_PORT && this.httpServer) {
+      this.ensureHttpServerExists(port, this.httpServer);
+      const wsServer = this.bindErrorHandler(
+        new wsPackage.Server({
+          noServer: true,
+          ...wsOptions,
+        }),
+      );
+
+      this.addWsServerToRegistry(wsServer, port, options.path || '/');
+      return wsServer;
+    }
+
+    if (server) {
+      return server;
+    }
+    if (options.path && port !== UNDERLYING_HTTP_SERVER_PORT) {
+      // Multiple servers with different paths
+      // sharing a single HTTP/S server running on different port
+      // than a regular HTTP application
+      const httpServer = this.ensureHttpServerExists(port);
+      httpServer?.listen(port);
+
+      const wsServer = this.bindErrorHandler(
+        new wsPackage.Server({
+          noServer: true,
+          ...wsOptions,
+        }),
+      );
+      this.addWsServerToRegistry(wsServer, port, options.path);
+      return wsServer;
+    }
+    const wsServer = this.bindErrorHandler(
+      new wsPackage.Server({
+        port,
+        ...wsOptions,
+      }),
+    );
+    return wsServer;
   }
 
-  bindMessageHandlers(
-    client: WebSocket,
+  public bindMessageHandlers(
+    client: any,
     handlers: MessageMappingProperties[],
-    process: (data: any) => Observable<any>,
+    transform: (data: any) => Observable<any>,
   ) {
-    fromEvent(client, 'message')
-      .pipe(
-        mergeMap((data) => this.bindMessageHandler(data, handlers, process)),
-        filter((result) => result),
-      )
-      .subscribe((response) => client.send(JSON.stringify(response)));
+    const close$ = fromEvent(client, CLOSE_EVENT).pipe(share(), first());
+    const source$ = fromEvent(client, 'message').pipe(
+      mergeMap((data) =>
+        this.bindMessageHandler(data, handlers, transform).pipe(
+          filter((result) => result),
+        ),
+      ),
+      takeUntil(close$),
+    );
+    const onMessage = (response: any) => {
+      if (client.readyState !== READY_STATE.OPEN_STATE) {
+        return;
+      }
+      client.send(JSON.stringify(response));
+    };
+    source$.subscribe(onMessage);
   }
 
-  bindMessageHandler(
-    buffer,
+  public bindMessageHandler(
+    buffer: any,
     handlers: MessageMappingProperties[],
-    process: (data: any) => Observable<any>,
+    transform: (data: any) => Observable<any>,
   ): Observable<any> {
-    const message = JSON.parse(buffer.data);
-    const messageHandler = handlers.find(
-      (handler) => handler.message === message.event,
-    );
-    if (!messageHandler) {
+    try {
+      const message = JSON.parse(buffer.data);
+      const messageHandler = handlers.find(
+        (handler) => handler.message === message.id,
+      );
+      const { callback } = messageHandler;
+      return transform(callback(message.data));
+    } catch {
       return EMPTY;
     }
-    console.log('tt', message.data);
-    return process(messageHandler.callback(message.data));
   }
 
-  close(server) {
-    server.close();
+  public bindErrorHandler(server: any) {
+    server.on(CONNECTION_EVENT, (ws: any) =>
+      ws.on(ERROR_EVENT, (err: any) => this.logger.error(err)),
+    );
+    server.on(ERROR_EVENT, (err: any) => this.logger.error(err));
+    return server;
+  }
+
+  public bindClientDisconnect(client: any, callback: Function) {
+    client.on(CLOSE_EVENT, callback);
+  }
+
+  public async dispose() {
+    const closeEventSignals = Array.from(this.httpServersRegistry)
+      .filter(([port]) => port !== UNDERLYING_HTTP_SERVER_PORT)
+      .map(([_, server]) => new Promise((resolve) => server.close(resolve)));
+
+    await Promise.all(closeEventSignals);
+    this.httpServersRegistry.clear();
+    this.wsServersRegistry.clear();
+  }
+
+  protected ensureHttpServerExists(
+    port: number,
+    httpServer = http.createServer(),
+  ) {
+    if (this.httpServersRegistry.has(port)) {
+      return;
+    }
+    this.httpServersRegistry.set(port, httpServer);
+
+    httpServer.on('upgrade', (request, socket, head) => {
+      const baseUrl = 'ws://' + request.headers.host + '/';
+      const pathname = new URL(request.url, baseUrl).pathname;
+      const wsServersCollection = this.wsServersRegistry.get(port);
+
+      let isRequestDelegated = false;
+      for (const wsServer of wsServersCollection) {
+        if (pathname === wsServer.path) {
+          wsServer.handleUpgrade(request, socket, head, (ws: unknown) => {
+            wsServer.emit('connection', ws, request);
+          });
+          isRequestDelegated = true;
+          break;
+        }
+      }
+      if (!isRequestDelegated) {
+        socket.destroy();
+      }
+    });
+    return httpServer;
+  }
+
+  protected addWsServerToRegistry<T extends Record<'path', string> = any>(
+    wsServer: T,
+    port: number,
+    path: string,
+  ) {
+    const entries = this.wsServersRegistry.get(port) ?? [];
+    entries.push(wsServer);
+
+    wsServer.path = path;
+    this.wsServersRegistry.set(port, entries);
   }
 }

ws/HeartBeat.js

@@ -0,0 +1,27 @@
+export default class Heartbeat {
+  constructor(e) {
+    E(this, '_interval', null);
+    E(this, 'ping', () => {
+      const e = Date.now().toString();
+      this.handler.ping(e);
+    });
+    this.handler = e;
+  }
+  ping() {
+    const e = Date.now().toString();
+    this.handler.ping(e);
+  }
+  start() {
+    this.stop(),
+      logger.debug(`Setting ping interval to ${PING_INTERVAL_MS}ms`),
+      (this._interval = window.setInterval(this.ping, PING_INTERVAL_MS));
+  }
+  stop() {
+    logger.debug('stop heartbeat'),
+      this._interval && window.clearInterval(this._interval);
+  }
+  pong(e, t) {
+    !e ||
+      (typeof e == 'string' && this.handler.pong(Date.now() - Number(e), t));
+  }
+}

ws/index.html

@@ -0,0 +1,24 @@
+<!DOCTYPE html>
+<html>
+  <head>
+    <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
+
+    <title>test socket</title>
+  </head>
+  <body>
+    <script type="module">
+      const socket = new WebSocket('ws://localhost:3100/ws?id=1111');
+      socket.addEventListener('message', (event) => {
+        console.log('event', event);
+      });
+      const test_message = {
+        data: JSON.stringify({ is_mobile: true }),
+        id: 'init_webrtc',
+      };
+      socket.onopen = (event) => {
+        console.log('onopen', event);
+        socket.send(JSON.stringify(test_message));
+      };
+    </script>
+  </body>
+</html>

ws/test.html

@@ -1,12 +0,0 @@
-<html>
-  <title>test ws</title>
-
-  <body>
-    <script type="module">
-        const 
-
-
-
-    </script>
-  </body>
-</html>