meta-server/src/meta.gateway.ts

import {
  SubscribeMessage,
  WebSocketGateway,
  OnGatewayInit,
  WebSocketServer,
  OnGatewayConnection,
  OnGatewayDisconnect,
} from '@nestjs/websockets';

import { Server } from 'ws';
import * as WebSocket from 'ws';

import {
  PeerConnection,
  initLogger,
  DataChannel,
  cleanup,
} from 'node-datachannel';

import { Buffer } from 'buffer';
import { Logger } from '@nestjs/common';
import * as path from 'path';
import { createReadStream } from 'fs';
import { SceneService } from './scene/scene.service';
// 'Verbose' | 'Debug' | 'Info' | 'Warning' | 'Error' | 'Fatal';
initLogger('Debug');

@WebSocketGateway({
  transports: ['websocket'],
  cors: '*',
  // namespace: "ws",
  path: '/ws',
})
export class MetaGateway
  implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect
{
  constructor(private readonly sceneService: SceneService) {}
  private logger: Logger = new Logger('MetaGateway');
  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) {
  //   this.logger.log(`payload: ${JSON.stringify(payload)}`);
  // }
  afterInit(server: Server) {
    this.logger.log('Init MetaGateway');
  }

  @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: '',
    };

    return pong;
  }

  @SubscribeMessage('init_webrtc')
  handleInitWebRtc(client: any, payload: any): void {
    console.log('handleInitWebRtc');
    console.log('this.sceneService', this.sceneService);
    const stun_server: string[] = Array.from(
      String(process.env.STUNS_SEVER).split(','),
    );
    this.logger.log('stun_server', stun_server);
    this.peer = new PeerConnection('roomTest', {
      portRangeBegin: 52000,
      portRangeEnd: 53000,
      iceServers: stun_server,
    });

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

    const replaceToPublic = (candidate) => {
      console.warn('PRIVATE_IP', process.env.PRIVATE_IP);
      return candidate.replace(process.env.PRIVATE_IP, process.env.PUBLIC_IP);
    };

    this.peer.onLocalCandidate((candidate, mid) => {
      if (/172\./.test(candidate)) {
        console.error('private Ip process', candidate);

        if (candidate.includes(process.env.PRIVATE_IP)) {
          console.error('PRIVATE_IP', process.env.PRIVATE_IP);
          candidate = replaceToPublic(candidate);
        } else {
          return;
        }
      }

      if (/192.168\./.test(candidate)) {
        if (!/192.168.0\./.test(candidate)) {
          console.warn('不是192.168.0.测试网段', candidate);
          return;
        }
        // if (candidate.includes(process.env.PRIVATE_IP)) {
        //   console.error('PRIVATE_IP', process.env.PRIVATE_IP);
        //   candidate = replaceToPublic(candidate);
        // }
      }
      console.warn('onLocalCandidate last 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.peer.onTrack((track) => {
      console.log('track', track);
    });

    this.gameChanel = this.peer.createDataChannel('game-input');

    this.peer.onDataChannel((dc) => {
      console.log('onDataChannel', dc);
    });

    this.gameChanel.onOpen(() => {
      console.log('channel is open');
      clearInterval(this.timer);

      const peers = this.peer.getSelectedCandidatePair();
      this.logger.log('配对成功', peers);

      let i = 1;
      const paths = path.join(__dirname, '../ws/video/100');
      console.error('__dirname', __dirname);
      console.error('paths', paths);
      if (this.gameChanel.isOpen()) {
        console.log('gameChanel', this.gameChanel.isOpen());
        this.sendWertcHeartPack(this.gameChanel);
      }

      Number.prototype.padLeft = function (n, str) {
        return Array(n - String(this).length + 1).join(str || '0') + this;
      };
      this.timer = setInterval(() => {
        if (i < 30) {
          const steam = createReadStream(
            paths + `/100.${Number(i).padLeft(4, '0')}.h264`,
          );
          // const steam = createReadStream(paths + `/test2`);
          steam.on('data', (data: Buffer) => {
            // console.log(data.buffer);
            const frame = new DataView(data.buffer);
            frame.setUint32(0, 1437227610);
            // frame.setUint16(6, 36);
            // frame.setUint16(24, 0);
            // frame.setUint16(26, 0);
            // frame.setUint32(28, 0);
            this.gameChanel.sendMessageBinary(Buffer.from(frame.buffer));
          });
        }
        i++;
      }, 1000 / 30);
    });
    this.gameChanel.onClosed(() => {
      console.log('gameChanel close');
      this.stopSendWertcHeartPack();
      cleanup();
    });
    this.gameChanel.onMessage((event) => {
      console.log('gameChanel onMessage', event);
      this.sceneService.handleMessage(event);
    });
    this.gameChanel.onError(() => {
      console.log('gameChanel onError');
      this.stopSendWertcHeartPack();
    });
  }

  sendWertcHeartPack(channel: DataChannel) {
    const heartPack = new DataView(new ArrayBuffer(4));
    heartPack.setUint32(0, 2009889916);
    this._webrtcInterval = setInterval(() => {
      if (channel.isOpen()) {
        channel.sendMessageBinary(Buffer.from(heartPack.buffer));
      }
    }, 200);
  }

  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.warn('收到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);
    this.sceneService.init(payload);
    this.logger.log(
      'start and send to gprc sceneService,method=>init',
      payload,
    );
  }

  handleConnection(client: WebSocket, ...args: any[]) {
    const { url } = args[0];
    console.log('url', url);
    this.logger.log(`Client connected:`);
    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);

    client.send(tt);
  }
  handleDisconnect(client: WebSocket) {
    this.logger.log(`Client disconnected: ${client.id}`);
    this.peer && this.peer.close();
  }
}