meta-server/src/scene/scene.service.ts

import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { ClientGrpc, Client } from '@nestjs/microservices';
import { grpcClientOptions } from './grpc-scene.options';
import { Logger } from '@nestjs/common';
import { DataChannel } from 'node-datachannel';
import { BehaviorSubject } from 'rxjs';
// import * as streamBuffers from 'stream-buffers';
import { ActionType } from './actionType';
import { CacheService } from 'src/cache/cache.service';
import { StreamService } from './stream/stream.service';
// import { InjectQueue } from '@nestjs/bull';
// import { Queue } from 'bull';
import { RotateService } from 'src/rotate/rotate.service';
import { DelayQueue, RxQueue, ThrottleQueue, DebounceQueue } from 'rx-queue';
import { MoveService } from 'src/move/move.service';
import { GetRouterService } from 'src/get-router/get-router.service';
import { join } from 'path';

@Injectable()
export class SceneService implements OnModuleInit, OnModuleDestroy {
  constructor(
    private cacheService: CacheService,
    private streamService: StreamService,
    private rotateService: RotateService,
    private moveService: MoveService,
    private getRouterService: GetRouterService, // @InjectQueue('rotate') private rotateQueue: Queue, // @InjectQueue('walking') private walkingQueue: Queue,
  ) { }
  @Client(grpcClientOptions) private readonly client: ClientGrpc;

  public _frameInteval: NodeJS.Timeout;
  public _frameTimeout: NodeJS.Timeout;
  public _rotateTimeout: NodeJS.Timeout;
  public _moveTimeout: NodeJS.Timeout;
  public startSteaming = new BehaviorSubject<boolean>(false);
  public onRotating = new BehaviorSubject<boolean>(false);
  public onMoving = new BehaviorSubject<boolean>(false);
  public frameCnt = new BehaviorSubject<number>(-1);
  private rotateframeCnt = -1;
  private moveframeCnt = -1;

  private sceneGrpcService: SceneGrpcService;
  private channel: DataChannel;
  private logger: Logger = new Logger('SceneService');
  private frameCntInterval = 1000;
  private user_id: string;
  private roomId: string;
  private onSteaming = false;

  private mockserverTime = Date.now() - 1653000000478;
  private lastRenderMedia = '';
  private frameCntSubscription: any;

  private roQueueSubscription: any;
  private moveQueueSubscription: any;
  private walkingSub: any;
  private joystickSub: any;

  private streamServiceSub: any;
  private roQueue: RxQueue = new DelayQueue(40);
  private clickQueue: RxQueue = new DebounceQueue(500);
  private moveQueue: RxQueue = new DelayQueue(10);
  private joystickQueue: RxQueue = new DebounceQueue(500);
  private rotateTimeStamp: number;
  private lastMoveCnt = -1;

  private firstRender = false;
  private latestBreakPointId: number;
  private isHoldingStream = false;

  public lastMoveStreamFrame = new BehaviorSubject<StreamFrameType>({
    frame: -1,
    clipPath: '',
    metaData: '',
  });
  public users = {};

  initUsers(app_id, userId) {
    const user = {
      appId: null,
      userId: null,
      breakPointId: null,
      roomId: null,
      player: {
        position: { x: -700, y: 0, z: 0 },
        angle: {
          pitch: 0,
          yaw: 0,
          roll: 0,
        },
      },
      camera: {
        position: { x: -1145, y: 0, z: 160 },
        angle: {
          pitch: 0,
          yaw: 0,
          roll: 0,
        },
      },
      rotateInfo: {
        frameIndex: 0,
        horizontal_move: 0,
        mediaSrc: null,
      },
      moveInfo: {},
      // traceIds: [],
      // actionResponses:[]
    };

    user.appId = app_id;
    user.userId = userId;
    user.breakPointId = 100;
    this.users[userId] = user;
  }

  onModuleInit(): void {
    this.sceneGrpcService =
      this.client.getService<SceneGrpcService>('SceneGrpcService');
    this.logger.log('init SceneGrpcService');
    this.streamServiceSub = this.streamService.onSteaming.subscribe((val) => {
      this.onSteaming = val;
    });
    Number.prototype.padLeft = function (n, str) {
      return Array(n - String(this).length + 1).join(str || '0') + this;
    };
  }

  public getConfig() {
    return {
      userId: this.user_id,
      roomId: this.roomId,
    };
  }

  public startStream(): void {
    clearInterval(this._frameInteval);
    if (this.frameCnt.value === -1) {
      this._frameInteval = setInterval(async () => {
        const next = this.frameCnt.value + 1;
        this.frameCnt.next(next);
      }, 1000);
    }
  }

  public holdSteam(): void {
    clearInterval(this._frameInteval);
    this.isHoldingStream = true;
  }

  public resumeStream(): void {
    this.onMoving.next(false);
    this.isHoldingStream = false;
    this.moveframeCnt = -1;
    this.rotateframeCnt = -1;
    clearInterval(this._frameInteval);
    this._frameInteval = setInterval(async () => {
      const next = this.frameCnt.getValue() + 1;
      this.frameCnt.next(next);
    }, 1000);
  }

  public stopStream(): void {
    if (this.frameCntSubscription) {
      this.frameCntSubscription.unsubscribe();
      this.frameCntSubscription = null;
    }
    if (this.roQueueSubscription) {
      this.roQueueSubscription.unsubscribe();
      this.roQueueSubscription = null;
    }
    if (this.moveQueueSubscription) {
      this.moveQueueSubscription.unsubscribe();
      this.moveQueueSubscription = null;
    }
    this.frameCnt.next(-1);
    clearInterval(this._frameInteval);
    this.rotateframeCnt = -1;
  }

  setConfig(user_id: string, roomId: string): void {
    this.user_id = user_id;
    this.roomId = roomId;
  }

  onModuleDestroy() {
    if ('unsubscribe' in this.streamServiceSub) {
      this.streamService.onSteaming.unsubscribe();
    }
  }

  init(request: InitRequest) {
    try {
      // const initReply = this.sceneGrpcService.init(request);
      // initReply.subscribe((reply) => {
      //   console.log('initReply', reply);
      // });
      this.rotateService.init(request.app_id, request.user_id);
      // this.moveService.init(request.app_id, request.user_id);

      // this.initUsers(request.app_id, request.user_id);
    } catch (error) {
      console.log('error', error);
    }
  }

  exit(request: ExitRequest) {
    this.frameCnt.next(-1);
    this.stopStream();
    // const exitReply = this.sceneGrpcService.exit(request);
    // exitReply.subscribe((reply) => {
    //   console.log('exitReply', reply);
    // });
  }

  async rotate(request: RotateRequest) {
    try {
      if (this.firstRender) {
        if (!this.roQueueSubscription) {
          this.handleRotateStream();
        }
        let redisMeta: StreamReplyType;
        this.onRotating.next(true);
        if (this.onMoving.value) {
          this.onMoving.next(false);
          const lastStreamFrame = this.lastMoveStreamFrame.getValue();
          const metaData: StreamReplyType = JSON.parse(
            lastStreamFrame.metaData,
          ) as any as StreamReplyType;
          const newUserStates: NewUserStatesType = metaData.newUserStates.find(
            (item) => item.userId === this.user_id,
          );
          const trace_id = metaData.traceIds[0];
          const userId = newUserStates.userId;
          const breakPointId = metaData.breakPointId;
          const cameraAngle = newUserStates.playerState.camera.angle;
          const playerAngle = newUserStates.playerState.player.angle;
          console.log('stop-data', trace_id, userId, cameraAngle, cameraAngle);
          redisMeta = await this.moveService.stop(
            trace_id,
            userId,
            breakPointId,
            cameraAngle,
            playerAngle,
          );
          console.log('stop-redisMeta', redisMeta);
          // redisMeta = await this.rotateService.rotate(request);
        } else {
          redisMeta = await this.rotateService.rotate(request);
        }

        if (redisMeta && 'mediaSrc' in redisMeta) {
          const mediaSrc: string = redisMeta.mediaSrc || '';
          if (mediaSrc.length > 0) {
            const src = mediaSrc.split('?')[0];
            // 临时本地替换路经
            // src = src.replace('/10086/', '');
            // 判断不是同一条源时才推出
            if (src.length > 0) {
              // console.log('不同源');
              // this.frameCnt += 1;
              this.holdSteam();
              this.lastRenderMedia = src;
              const clipPath = join(__dirname, `../ws/${src}`);
              // console.log('src-clipPath', src, clipPath);
              delete redisMeta.mediaSrc;

              // const nextFrame = this.frameCnt.getValue() + 1;
              // console.log('nextFrame', nextFrame);
              // this.frameCnt.next(nextFrame);
              const random_boolean = Math.random() < 0.3;

              const stream: StreamFrameType = {
                frame: -1,
                clipPath: clipPath,
                metaData: JSON.stringify(redisMeta),
                serverTime: this.mockserverTime,
                DIR: this.frameCnt.getValue() < 10 ? 3 : random_boolean ? 1 : 3,
              };
              // this.rotateQueue.add(stream, {
              //   delay: 5,
              //   jobId: `rotate:${this.user_id}:${this.frameCnt.getValue()}`,
              //   removeOnComplete: true,
              // });
              clearTimeout(this._rotateTimeout);
              this.roQueue.next(stream);
            } else {
              // this.onRotating.next(false);
            }
          }
        }
      }
    } catch (error) {
      this.logger.error('rotate', error);
      console.log('error', error);
    }
  }

  async walking(req) {
    try {
      console.log('walking', req);
      this.clickQueue.next(req);
      this.walkingSub = this.clickQueue.subscribe(async (request) => {
        const user = this.moveService.users[this.user_id];
        const path = await this.getRouterService.searchRoad(
          user.appId,
          user.breakPointId,
          req.clicking_action.clicking_point,
        );
        const walkingRes = await this.moveService.move(path, request);

        // console.log('walkingRes-front', walkingRes);
        if (walkingRes && !this.onMoving.value) {
          // console.log('walkingRes-front', walkingRes);
          // shift出前第一个镜头数据
          const rotateCamData = walkingRes.shift();
          if (rotateCamData?.length) {
            rotateCamData.forEach((item: StreamReplyType) => {
              item.type = 'rotate';
            });
          }
          // walkingRes marker to everybody
          const seqs = Array.from(
            walkingRes,
          ).flat() as any as StreamReplyType[];

          if (seqs?.length) {
            const lastSeq = rotateCamData?.length
              ? (Array.from(rotateCamData).concat(
                seqs,
              ) as any as StreamReplyType[])
              : seqs;
            this.handleSeqMoving(lastSeq);
          } else {
            console.error('walking-move无数据');
          }
          // this.lastMoveCnt = this.frameCnt.value + seqs.length;
        }
      });
    } catch (error) {
      this.logger.error('walking', error);
    }
  }

  async joystick(request: JoystickRequest) {
    try {
      this.joystickQueue.next(request);
      if (!this.joystickSub) {
        this.joystickSub = this.joystickQueue.subscribe(
          async (joystickRequest) => {
            const joystickRes = await this.moveService.joystick(
              joystickRequest,
            );
            console.log('joystickRes-front', joystickRes);
            // 有数据 [0]是rotate数据，[1-infinity]是walking数据
            if (Array.isArray(joystickRes)) {
              // shift出前第一个镜头数据
              const rotateCamData = joystickRes.shift();
              console.log('rotateCamData', rotateCamData);
              if (rotateCamData?.length) {
                rotateCamData.forEach((item: StreamReplyType) => {
                  item.type = 'rotate';
                });
              }
              const seqs = Array.from(
                joystickRes,
              ).flat() as any as StreamReplyType[];
              if (seqs?.length || rotateCamData?.length) {
                const lastSeq = rotateCamData?.length
                  ? (Array.from(rotateCamData).concat(
                    seqs,
                  ) as any as StreamReplyType[])
                  : seqs;
                this.handleSeqMoving(lastSeq);
              } else {
                console.warn('joystick-move无数据');
              }
            } else {
              console.log('转交数据');
              this.streamService.pushNormalDataToStream(request);
            }
          },
        );
      }
    } catch (error) {
      this.logger.error('joystick', error);
    }
  }
  /**
   * 主要处理moving的序列动作
   * @param seqs StreamReplyType[]
   */
  handleSeqMoving(seqs: StreamReplyType[]) {
    if (!this.moveQueueSubscription) {
      this.handleMoveSteam();
    }
    console.log('moving-seqs', seqs.length);
    this.onMoving.next(true);
    this.holdSteam();

    seqs.forEach((frame: StreamReplyType, index:number) => {
      const mediaSrc = frame.mediaSrc;
      const src = mediaSrc.split('?')[0];
      // 临时本地替换路经
      // src = src.replace('//', '');
      const clipPath = join(__dirname, `../ws/${src}`);
      const type = frame.type?.length ? frame.type.slice() : 'move';

      console.log('type', frame.type);
      delete frame.mediaSrc;
      delete frame.type;
      //TODO 
      const random_boolean = Math.random() < 0.3;
      // const random_index === 1 ? 3 : random_boolean ? 1 : 3;
      const stream: StreamFrameType = {
        frame: -1,
        clipPath: clipPath,
        metaData: JSON.stringify(frame),
        serverTime: this.mockserverTime,
        DIR: 3,
        type: type,
      };
      this.moveQueue.next(stream);
    });
  }

  cleanMoveSteam() {
    if (this.moveQueueSubscription) {
      this.moveQueueSubscription.unsubscribe();
      this.lastMoveCnt = -1;
      this.moveQueueSubscription = null;
    }
    if (this.walkingSub) {
      this.walkingSub.unsubscribe();
      this.walkingSub = null;
    }
  }
  handleMoveSteam() {
    this.moveQueueSubscription = this.moveQueue.subscribe(
      async (stream: StreamFrameType) => {
        const metaData: StreamReplyType = JSON.parse(stream.metaData);

        if (this.moveframeCnt === -1) {
          this.moveframeCnt = this.frameCnt.value;
        }
        this.moveframeCnt += 1;
        this.latestBreakPointId = metaData.breakPointId;
        // if (this.onMoving) {
        //   this.frameCnt.next(this.moveframeCnt);
        // } else {
        //   console.log(
        //     'handleMoveSteam stop',
        //     this.moveframeCnt,
        //     this.currentMoveMaker,
        //   );
        //   this.cleanMoveSteam();
        //   this.resumeStream();
        //   return;
        // }
        const streamData: StreamFrameType = {
          frame: this.moveframeCnt,
          clipPath: stream.clipPath,
          metaData: stream.metaData,
          serverTime: this.mockserverTime,
          DIR: 3,
        };
        console.log(
          '[media-move]',
          this.moveframeCnt,
          stream.clipPath,
          stream.type,
        );

        this.lastMoveStreamFrame.next(streamData);
        const res = await this.streamService.pushFrameToSteam(streamData);
        if (res.done) {
          clearTimeout(this._moveTimeout);
          this._moveTimeout = setTimeout(() => {
            console.log('move 交权给空流,当前pts', res.frame);
            //TODO 每个结束点 updateUser metaData
            const lastFrame = this.lastMoveStreamFrame.getValue();
            const lastFrameMeta = JSON.parse(lastFrame.metaData);

            const userId = this.user_id;
            const breakPointId = lastFrameMeta.breakPointId;
            const lastReply = lastFrameMeta;
            this.moveService.updateUser(userId, breakPointId, lastReply);
            this.frameCnt.next(res.frame);
            this.resumeStream();
            this.rotateframeCnt = -1;
            this.onMoving.next(false);
            console.log('move end');
          }, 300);
        }
      },
    );
  }

  handleDataChanelOpen(channel: DataChannel): void {
    this.channel = channel;
    this.streamService.setChannel(channel);
    this.startSteaming.next(true);
    this.startStream();
    this.handleStream();
  }

  handleDataChanelClose(): void {
    this.stopStream();
    this.startSteaming.next(false);
    this.streamService.closeChannel();
    const exitRequest: ExitRequest = {
      action_type: 1002,
      user_id: this.user_id,
      trace_id: '',
    };
    this.exit(exitRequest);
  }

  handleMessage(message: string | Buffer) {
    try {
      if (typeof message === 'string') {
        // wasm:特例, requestIframe
        if (message.includes('wasm:')) {
          const msg: RTCMessageRequest = JSON.parse(
            message.replace('wasm:', ''),
          );
          if (msg.MstType === 0) {
            this.logger.log('lost I frame');
            this.handleIframeRequest();
          }
        } else {
          const msg: RTCMessageRequest = JSON.parse(message);
          switch (msg.action_type) {
            case ActionType.walk:
              const walk = msg;
              this.walking(walk);
              break;
            case ActionType.joystick:
              const JoystickRequest = msg as any as JoystickRequest;
              this.joystick(JoystickRequest);
              break;
            case ActionType.breathPoint:
              this.handleBreath(msg);
              break;
            case ActionType.rotate:
              const rotateRequest: RotateRequest = msg;
              this.rotate(rotateRequest);
              break;
            case ActionType.userStatus:
              this.updateUserStatus(msg);
              break;
            case ActionType.status:
              this.updateStatus();
              break;
            default:
              break;
          }
        }
      }
    } catch (error) {
      this.logger.error('handleMessage:rtc--error', message);
    }
  }

  async handleIframeRequest() {
    //TODO Iframe 最终传什么？
    // const lastStreamFrame = this.streamService.lastStreamFrame.getValue();
    // lastStreamFrame.DIR = 1;
    // console.log('lastStreamFrame', lastStreamFrame);
    // const nextFrame = this.frameCnt.getValue() + 1;
    // lastStreamFrame.frame = nextFrame;
    // this.frameCnt.next(nextFrame);
    // this.streamService.pushFrameToSteam(lastStreamFrame);
    // const redisDataAuto = await this.rotateService.echo(this.user_id);
    // if (redisDataAuto) {
    //   'mediaSrc' in redisDataAuto && delete redisDataAuto.mediaSrc;
    //   const streamMeta: StreamMetaType = {
    //     frame: nextFrame,
    //     metaData: JSON.stringify(redisDataAuto),
    //   };
    //   this.streamService.pushMetaDataToSteam(streamMeta);
    // }
  }

  async handleBreath(request) {
    const npsRes = await this.moveService.getBreakPoints(request);
    // console.log('npsRes', npsRes);
    this.streamService.pushNormalDataToStream(npsRes);
  }

  updateStatus() {
    const reply = {
      data: { action_type: 1009, echo_msg: { echoMsg: Date.now() } },
      track: false,
    };
    this.streamService.pushNormalDataToStream(reply);
  }
  async updateUserStatus(request) {
    try {
      const redisMeta = await this.rotateService.getNewUserStateRequest(
        request,
      );
      if (redisMeta) {
        redisMeta.actionType = 1024;
        this.streamService.pushNormalDataToStream(redisMeta);
      }
    } catch (error) {
      this.logger.error('updateUserStatus::function', error);
    }
  }

  pushFirstRender(clipPath: string, metaData: string): Promise<boolean> {
    return new Promise<boolean>(async (resolve, reject) => {
      try {
        const streamData: StreamFrameType = {
          frame: 1,
          clipPath: clipPath,
          metaData: metaData,
          serverTime: this.mockserverTime,
          DIR: 1,
        };
        const hasPush = await this.streamService.pushFrameToSteam(streamData);
        return resolve(hasPush.done);
      } catch (error) {
        return reject(false);
      }
    });
  }

  handleStream() {
    this.frameCntSubscription = this.frameCnt.subscribe(async (frame) => {
      try {
        console.log('frame', frame);
        if (frame === 1) {
          const redisData = await this.rotateService.echo(this.user_id);
          this.onSteaming = true;
          this.holdSteam();
          console.log('redisData', redisData);
          if (redisData && 'mediaSrc' in redisData) {
            const mediaSrc: string = redisData.mediaSrc || '';
            if (mediaSrc.length > 0) {
              const src = mediaSrc.split('?')[0];
              // 临时本地替换路经
              // src = src.replace('/10086/', '');
              const clipPath = join(__dirname, `../ws/${src}`);
              delete redisData.mediaSrc;
              this.logger.log(
                `user:${this.user_id}:first render stream` +
                JSON.stringify({ path: clipPath, meta: redisData }),
              );
              const status = await this.pushFirstRender(
                clipPath,
                JSON.stringify(redisData),
              );
              if (status) {
                this.firstRender = true;
                this.frameCnt.next(2);
                this.resumeStream();
              } else {
                this.logger.error('first render problem', status);
              }
            }
          }
        }
        if (
          frame > 1 &&
          !this.onMoving.value &&
          !this.onRotating.value &&
          this.firstRender
        ) {
          const redisDataAuto = await this.rotateService.echo(this.user_id);
          if (redisDataAuto) {
            console.log(`空白流::有数据：${frame}`);
            'mediaSrc' in redisDataAuto && delete redisDataAuto.mediaSrc;
            const streamMeta: StreamMetaType = {
              frame: frame,
              metaData: JSON.stringify(redisDataAuto),
            };
            this.streamService.pushMetaDataToSteam(streamMeta);
          } else {
            this.stopStream();
            console.log('空流无Redis数据');
            throw new Error('空流无Redis数据');
          }
        }
      } catch (error) {
        this.stopStream();
        this.logger.error('handleStream', error);
      }
    });
  }
  handleRotateStream() {
    this.roQueueSubscription = this.roQueue.subscribe(
      async (stream: StreamFrameType) => {
        this.rotateTimeStamp = Date.now();
        if (this.rotateframeCnt === -1) {
          this.rotateframeCnt = this.frameCnt.value;
        }
        this.rotateframeCnt += 1;

        stream.frame = this.rotateframeCnt;
        console.log(
          '[media-rotate]',
          stream.frame,
          this.rotateframeCnt,
          stream.clipPath,
        );
        // this.logger.log(
        //   `roQueueSubscription:frame:${this.rotateframeCnt}  ` +
        //   JSON.stringify(stream.metaData),
        // );

        const res = await this.streamService.pushFrameToSteam(stream);
        if (res.done) {
          clearTimeout(this._rotateTimeout);
          this._rotateTimeout = setTimeout(() => {
            console.log('rotate 1秒', Date.now());
            console.log('rotate end');
            // const next = res.frame + 1;
            this.frameCnt.next(res.frame);
            this.resumeStream();
            this.rotateframeCnt = -1;
            this.onMoving.next(false);
            this.onRotating.next(false);
          }, 300);
        }
      },
    );
  }
}