update

config.yaml

@@ -3,31 +3,31 @@ http:
   port: 6688
 
 
-grpc:
-  url: '221.4.210.172:23000'
 # grpc:
-#   url: '192.168.0.47:3000'
+#   url: '221.4.210.172:23000'
+grpc:
+  url: '192.168.0.200:3000'
+
+# redis:
+#   port: 26379
+#   host: '221.4.210.172' #远程调试需要设置bindip 为0.0.0.0 并且设置密码
+#   password: '' # 非远程不需要密码
+#   decode_responses: true
+#   db: 9
 
 redis:
-  port: 26379
-  host: '221.4.210.172' #远程调试需要设置bindip 为0.0.0.0 并且设置密码
+  port: 6379
+  host: '192.168.0.47' #远程调试需要设置bindip 为0.0.0.0 并且设置密码
   password: '' # 非远程不需要密码
   decode_responses: true
   db: 9
 
 queueRedis:
-  port: 26379
-  host: '221.4.210.172' #远程调试需要设置bindip 为0.0.0.0 并且设置密码
-  password: '' # 非远程不需要密码
+  port: 6379
+  host: 'localhost' #远程调试需要设置bindip 为0.0.0.0 并且设置密码
+  password: 'redis9394' # 非远程不需要密码
   decode_responses: true
-  db: 14
-
-# redis:
-#   port: 6379
-#   host: '192.168.0.47' #远程调试需要设置bindip 为0.0.0.0 并且设置密码
-#   password: '' # 非远程不需要密码
-#   decode_responses: true
-#   db: 9
+  db: 0
 
 stun:
   server: ['stun:172.18.156.41:3478', 'stun:120.24.252.95:3478']
@@ -38,9 +38,9 @@ server:
   private_ip: 172.18.197.114
   public_ip: 120.24.252.95
 
-
+  
 # PRIVATE_IP=172.18.197.114
 # PUBLIC_IP=120.24.252.95
 # STUNS_SEVER="stun:172.18.156.41:3478,stun:120.24.252.95:3478"
 # GRPC_URL="192.168.0.48:3000"
-# REDIS_URL="redis://:192.168.0.47:6379/9"
+# REDIS_URL="redis://:192.168.0.47:6379/9"

src/move/move.service.ts

@@ -1,15 +1,20 @@
 import { Injectable } from '@nestjs/common';
 import { CacheService } from 'src/cache/cache.service';
+import { RotateService } from 'src/rotate/rotate.service';
+// import { SceneService } from 'src/scene/scene.service';
 
 @Injectable()
 export class MoveService {
-  constructor(private cacheService: CacheService) {}
+  constructor(
+    private cacheService: CacheService,
+    private rotateService: RotateService,
+  ) {}
   private Actions = {
     Clicking: 1,
     Rotation: 1014,
     Joystick: 15,
   };
-  public users = {};
+  public users = this.rotateService.users;
 
   private reply = {
     traceIds: [],
@@ -153,14 +158,22 @@ export class MoveService {
 
       //纠正，旋转传到缓存里
       const checkReplys = [];
+      // console.log('矫正： ' + angle + ' 度');
       for (let i = 0; i < angle; ++i) {
-        //let reply = rotateService.rotateForAngle(userId,(angle+i)%359);
-        //reply.traceIds = [];
-        //reply.traceIds.push(traceId);
-        //let actionResponse = rotateService.createActionResponse(actionType,trace_id);
-        //reply.actionResponses = [];
-        //reply.actionResponses.push(actionResponse);
-        //checkReplys.push(reply);
+        // console.warn('矫正一次：' + i);
+        const reply = await this.rotateService.rotateForAngle(userId, 1);
+        // console.warn(
+        //   '矫正：' + reply.newUserStates[0].playerState.camera.angle.yaw,
+        // );
+        reply.traceIds = [];
+        reply.traceIds.push(traceId);
+        const actionResponse = this.rotateService.createActionResponse(
+          actionType,
+          traceId,
+        );
+        reply.actionResponses = [];
+        reply.actionResponses.push(actionResponse);
+        checkReplys.push(reply);
       }
       replys['P' + user.breakPointId + 'T' + user.breakPointId] = checkReplys;
       //replys.push(checkReplys);
@@ -421,4 +434,141 @@ export class MoveService {
     reply['newUserStates'][0].renderInfo.isMoving = 0;
     return reply;
   }
+
+  async joystick(actionRequest) {
+    try {
+      const userId = actionRequest['user_id'];
+      const traceId = actionRequest['trace_id'];
+      const dir_action = actionRequest['dir_action'];
+      const user = this.users[userId];
+
+      const appId = user.appId;
+      //只是移动人物
+      if (dir_action.speed_level < 7) {
+        user.player.angle.yaw = dir_action.move_angle;
+        this.reply['newUserStates'][0]['userId'] = userId;
+        this.reply['newUserStates'][0].playerState.player.position =
+          user.player.position;
+        this.reply['newUserStates'][0].playerState.player.angle.yaw =
+          dir_action.move_angle;
+
+        this.reply['newUserStates'][0].playerState.camera.position =
+          user.camera.position;
+        this.reply['newUserStates'][0].playerState.camera.angle =
+          user.camera.angle;
+
+        this.reply['newUserStates'][0].playerState.cameraCenter =
+          user.camera.position;
+
+        this.reply['actionResponses'][0].traceId = traceId;
+
+        return this.reply;
+      }
+      //选择过渡
+      else {
+        const breakPointId = user.breakPointId;
+        //先矫正
+        const breakPointRes = await this.cacheService.get(
+          'breakpoints:app_id:' + appId + ':break_point_id:' + breakPointId,
+        );
+        if (breakPointRes == null) {
+          return null;
+        }
+
+        const breakPoint = JSON.parse(breakPointRes);
+        const contact = breakPoint.contact;
+        let chooseBreakPointId = null;
+        let minOffsetAngle = null;
+        let neighPoint = null;
+        let angle = 0;
+        for (let i = 0; i < contact.length; ++i) {
+          neighPoint = await this.cacheService.get(
+            'breakpoints:app_id:' + appId + ':break_point_id:' + contact[i],
+          ); //通过contact[i],去redis里找
+          //通过user.player.position;neighPoint.position获得角度
+          angle = this.getAngle(
+            user.player.position,
+            { x: user.player.position.x + 1, y: user.player.position.y },
+            neighPoint.position,
+          );
+          if (
+            Math.abs(angle - dir_action.move_angle) < 45 &&
+            (minOffsetAngle == null ||
+              Math.abs(angle - dir_action.move_angle) < minOffsetAngle)
+          ) {
+            chooseBreakPointId = contact[i];
+            minOffsetAngle = Math.abs(angle - dir_action.move_angle);
+          }
+        }
+
+        if (chooseBreakPointId == null) {
+          return null;
+        } else {
+          //人物矫正
+          user.player.angle.yaw = angle;
+
+          //相机纠正
+          const replys = [];
+          const traceIds = [];
+          traceIds.push(traceId);
+
+          const checkReplys = [];
+          angle = user.camera.angle.yaw % 45; //纠正需要
+          for (let i = 0; i < angle; ++i) {
+            //let reply = rotateService.rotateForAngle(userId,(angle+i)%359);
+            //reply.traceIds = [];
+            //reply.traceIds.push(traceId);
+            //let actionResponse = rotateService.createActionResponse(actionType,trace_id);
+            //reply.actionResponses = [];
+            //reply.actionResponses.push(actionResponse);
+            //checkReplys.push(reply);
+          }
+          replys.push(checkReplys);
+
+          //过渡
+          //读redis里的数据，按照frame_index的大小排序
+          const key =
+            'moveframe:app_id:' +
+            appId +
+            ':start_break_point_id:' +
+            breakPointId +
+            ':end_break_point_id:' +
+            chooseBreakPointId +
+            ':angle:' +
+            (user.camera.angle % 45);
+          const moveFramesRes = await this.cacheService.get(key);
+          if (moveFramesRes == null) {
+            return replys;
+          }
+          const moveFrames = JSON.parse(moveFramesRes);
+          //读redis里的数据
+          const startBreakPointRes = await this.cacheService.get(
+            'breakpoints:app_id:' + appId + ':break_point_id:' + breakPointId,
+          );
+          const startBreakPoint = JSON.parse(startBreakPointRes);
+
+          const endBreakPointRes = await this.cacheService.get(
+            'breakpoints:app_id:' +
+              appId +
+              ':break_point_id:' +
+              chooseBreakPointId,
+          );
+          const endBreakPoint = JSON.parse(endBreakPointRes);
+
+          const pathReplys = this.createCacheReplys(
+            moveFrames,
+            traceId,
+            userId,
+            breakPointId,
+            startBreakPoint.position,
+            endBreakPoint.position,
+          );
+          replys.push(pathReplys);
+          return replys;
+        }
+      }
+    } catch (error) {
+      console.log('MoveService', error);
+    }
+  }
 }

src/rotate/rotate.service.spec.ts

@@ -1,18 +0,0 @@
-import { Test, TestingModule } from '@nestjs/testing';
-import { RotateService } from './rotate.service';
-
-describe('RotateService', () => {
-  let service: RotateService;
-
-  beforeEach(async () => {
-    const module: TestingModule = await Test.createTestingModule({
-      providers: [RotateService],
-    }).compile();
-
-    service = module.get<RotateService>(RotateService);
-  });
-
-  it('should be defined', () => {
-    expect(service).toBeDefined();
-  });
-});

src/rotate/rotate.service.ts

@@ -1,9 +1,12 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { CacheService } from 'src/cache/cache.service';
+// import { SceneService } from 'src/scene/scene.service';
 
 @Injectable()
 export class RotateService {
-  constructor(private cacheService: CacheService) {}
+  constructor(
+    private cacheService: CacheService, // private sceneService: SceneService,
+  ) {}
   private actionRequestPool = {};
   private logger: Logger = new Logger('rotateService');
   private Actions = {
@@ -11,7 +14,7 @@ export class RotateService {
     Rotation: 1014,
     Joystick: 15,
   };
-  private users = {};
+  public users = {};
   private replies = {};
 
   init(app_id, userId) {
@@ -176,7 +179,7 @@ export class RotateService {
       }
 
       actionRequests.splice(0, sub);
-      const hAngle = horizontal_move * 90;
+      const hAngle = horizontal_move * 60;
       if (Math.abs(hAngle) < 1) {
         user.rotateInfo.horizontal_move = horizontal_move;
         //user.traceIds = traceIds;
@@ -243,11 +246,11 @@ export class RotateService {
   async rotateForAngle(userId, hAngle) {
     try {
       const user = this.users[userId];
-      user.rotateInfo.frameIndex += Math.floor(hAngle);
-      if (user.rotateInfo.frameIndex < 0) {
-        user.rotateInfo.frameIndex = 360 - user.rotateInfo.frameIndex;
-      } else if (user.rotateInfo.frameIndex > 359) {
-        user.rotateInfo.frameIndex -= 360;
+      user.camera.angle.yaw += Math.floor(hAngle);
+      if (user.camera.angle.yaw < 0) {
+        user.camera.angle.yaw = 360 + user.camera.angle.yaw;
+      } else if (user.camera.angle.yaw > 359) {
+        user.camera.angle.yaw -= 360;
       }
 
       const reply = JSON.parse(JSON.stringify(this.replies[userId]));
@@ -258,17 +261,17 @@ export class RotateService {
         'rotateframe:app_id:' +
         user.appId +
         ':frame_index:' +
-        user.rotateInfo.frameIndex +
+        user.camera.angle.yaw +
         ':break_point_id:' +
         user.breakPointId;
       // const value = null;
-
+      console.log('矫正：'+key);
       const redisData = await this.cacheService.get(key);
       if (redisData && redisData.length > 0) {
         const value = redisData ? JSON.parse(redisData) : null;
         // console.log('rotate-service', value);
         user.camera['position'] = value ? value.cameraPosition : '';
-        user.camera['angle'] = value ? value.cameraAngle : '';
+        //user.camera['angle'] = value ? value.cameraAngle : '';
 
         reply['newUserStates'][0]['playerState'].player.position =
           user.player.position;

src/scene/rotate-consumer.ts

@@ -20,7 +20,7 @@ export class RotateConsumer {
     private streamService: StreamService,
     private sceneService: SceneService,
   ) {}
- 
+  private isDone = true;
   private _checkerRotateDone: NodeJS.Timeout;
   @Process()
   async processFrame(job: Job<unknown>) {

src/scene/scene.module.ts

@@ -32,7 +32,7 @@ import { GetRouterService } from 'src/get-router/get-router.service';
     MoveService,
     GetRouterService,
   ],
-  exports: [SceneService, CacheService],
+  exports: [SceneService, CacheService, RotateService, MoveService],
 })
 export class SceneModule implements OnModuleInit {
   onModuleInit() {

src/scene/scene.service.ts

@@ -49,24 +49,65 @@ export class SceneService implements OnModuleInit, OnModuleDestroy {
   private mockserverTime = Date.now() - 1653000000478;
   private lastRenderMedia = '';
   private frameCntSubscription: any;
-  // private roQueueSubscription: any;
-  // private moveQueueSubscription: any;
+
+  private roQueueSubscription: any;
+  private moveQueueSubscription: any;
   private walkingSub: any;
 
   private streamServiceSub: any;
-
+  private roQueue: RxQueue = new DelayQueue(100);
   private clickQueue: RxQueue = new DebounceQueue(500);
-  private moveQueue: RxQueue = new DelayQueue(100);
+  private moveQueue: RxQueue = new DelayQueue(600);
   private rotateTimeStamp: number;
   private lastMoveCnt = -1;
 
   private firstRender = false;
+  private currentMoveMaker = '';
 
   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 =
@@ -80,6 +121,13 @@ export class SceneService implements OnModuleInit, OnModuleDestroy {
     };
   }
 
+  public getConfig() {
+    return {
+      userId: this.user_id,
+      roomId: this.roomId,
+    };
+  }
+
   public startStream(): void {
     clearInterval(this._frameInteval);
     if (this.frameCnt.value === -1) {
@@ -110,14 +158,14 @@ export class SceneService implements OnModuleInit, OnModuleDestroy {
       this.frameCntSubscription.unsubscribe();
       this.frameCntSubscription = null;
     }
-    // if (this.roQueueSubscription) {
-    //   this.roQueueSubscription.unsubscribe();
-    //   this.roQueueSubscription = null;
-    // }
-    // if (this.moveQueueSubscription) {
-    //   this.moveQueueSubscription.unsubscribe();
-    //   this.moveQueueSubscription = 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;
@@ -148,7 +196,11 @@ export class SceneService implements OnModuleInit, OnModuleDestroy {
       //   console.log('initReply', reply);
       // });
       this.rotateService.init(request.app_id, request.user_id);
-      this.moveService.init(request.app_id, request.user_id);
+      // this.moveService.init(request.app_id, request.user_id);
+
+      // this.initUsers(request.app_id, request.user_id);
+      this.rotateQueue.empty();
+      this.walkingQueue.empty();
     } catch (error) {
       console.log('error', error);
     }
@@ -211,23 +263,24 @@ export class SceneService implements OnModuleInit, OnModuleDestroy {
             const clipPath = join(__dirname, `../ws/video/${src}`);
             // console.log('src-clipPath', src, clipPath);
             delete redisMeta.mediaSrc;
-            if (this.rotateframeCnt === -1) {
-              this.rotateframeCnt = this.frameCnt.value;
-            }
-            this.rotateframeCnt += 1;
-            const random_boolean = Math.random() < 0.5;
+            // if (this.rotateframeCnt === -1) {
+            //   this.rotateframeCnt = this.frameCnt.value;
+            // }
+            const nextFrame = this.frameCnt.getValue() + 1;
+            this.frameCnt.next(nextFrame);
+            const random_boolean = Math.random() < 0.3;
 
             const stream: StreamFrameType = {
-              frame: this.rotateframeCnt,
+              frame: this.frameCnt.getValue(),
               clipPath: clipPath,
               metaData: JSON.stringify(redisMeta),
               serverTime: this.mockserverTime,
               DIR: random_boolean ? 1 : 3,
             };
             this.rotateQueue.add(stream, {
-              delay: 5,
-              jobId: `rotate:${this.user_id}:${this.rotateframeCnt}`,
-              // lifo: true,
+              delay: 10,
+              jobId: `rotate:${this.user_id}:${this.frameCnt.getValue()}`,
+              removeOnComplete: true,
             });
           } else {
             this.onRotating.next(false);
@@ -308,22 +361,23 @@ export class SceneService implements OnModuleInit, OnModuleDestroy {
   }
 
   async handleIframeRequest() {
-    // const lastStreamFrame = this.streamService.lastStreamFrame.getValue();
-    // lastStreamFrame.DIR = 1;
-    // console.log('lastStreamFrame', lastStreamFrame);
+    const lastStreamFrame = this.streamService.lastStreamFrame.getValue();
+    lastStreamFrame.DIR = 1;
+    console.log('lastStreamFrame', lastStreamFrame);
     const nextFrame = this.frameCnt.getValue() + 1;
-    // lastStreamFrame.frame = 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);
-    }
+    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 walking(req) {
@@ -338,49 +392,63 @@ export class SceneService implements OnModuleInit, OnModuleDestroy {
           req.clicking_action.clicking_point,
         );
         const walkingRes = await this.moveService.move(path, request);
-        // console.log('walkingRes', walkingRes);
+        // this.moveService.rotateForAngle();
+        console.log('walkingRes', walkingRes);
+        // debugger;
+
         if (walkingRes && !this.onMoving.value) {
-          this.onMoving.next(true);
-          this.holdSteam();
-          // if (!this.moveQueueSubscription) {
-          //   this.handleMoveSteam();
-          // }
-          if (this.moveframeCnt === -1) {
-            this.moveframeCnt = this.frameCnt.value;
-          }
 
-          const res = Object.keys(walkingRes).map((item) => {
-            console.log('item', item);
-            return Array.from(walkingRes[item]).map((i) => {
-              i['marker'] = item;
-              return i;
-            });
-          });
+          // walkingRes marker to everybody
+          const res: ArrayLike<unknown> = Object.keys(walkingRes).map(
+            (item) => {
+              console.log('item', item);
+              return Array.from(walkingRes[item]).map((i) => {
+                i['marker'] = item;
+                return i;
+              });
+            },
+          );
+          //
           const seqs = Array.from(res).flat();
-          // this.lastMoveCnt = this.frameCnt.value + seqs.length;
-          seqs.forEach((frame: StreamReplyType) => {
-            const mediaSrc = frame.mediaSrc;
-            let src = mediaSrc.split('?')[0];
-            // 临时本地替换路经
-            src = src.replace('/0000000001/', '');
-            const clipPath = join(__dirname, `../ws/video/${src}`);
-            this.moveframeCnt += 1;
-            delete frame.mediaSrc;
-            const stream: StreamFrameType = {
-              frame: this.moveframeCnt,
-              clipPath: clipPath,
-              metaData: JSON.stringify(frame),
-              serverTime: this.mockserverTime,
-              DIR: 1,
-            };
 
-            this.walkingQueue.add(stream, {
-              delay: 5,
-              jobId: `walking:${this.user_id}:${this.moveframeCnt}`,
-              // lifo: true,
+          if (seqs?.length) {
+            if (!this.moveQueueSubscription) {
+              this.handleMoveSteam();
+            }
+            console.log('walking-seqs', seqs);
+            this.onMoving.next(true);
+            this.holdSteam();
+            // if (this.moveframeCnt === -1) {
+            //   this.moveframeCnt = this.frameCnt.value;
+            // }
+            this.moveframeCnt = this.frameCnt.value;
+            seqs.forEach((frame: StreamReplyType) => {
+              const mediaSrc = frame.mediaSrc;
+              let src = mediaSrc.split('?')[0];
+              // 临时本地替换路经
+              src = src.replace('/0000000001/', '');
+              const clipPath = join(__dirname, `../ws/video/${src}`);
+              this.moveframeCnt += 1;
+              delete frame.mediaSrc;
+              const stream: StreamFrameType = {
+                frame: this.moveframeCnt,
+                clipPath: clipPath,
+                metaData: JSON.stringify(frame),
+                serverTime: this.mockserverTime,
+                DIR: 1,
+              };
+
+              // this.walkingQueue.add(stream, {
+              //   delay: 10,
+              //   jobId: `walking:${this.user_id}:${this.moveframeCnt}`,
+              //   // lifo: true,
+              // });
+              this.moveQueue.next(stream);
             });
-            // this.moveQueue.next(stream);
-          });
+          } else {
+            console.log('walking无数据');
+          }
+          // this.lastMoveCnt = this.frameCnt.value + seqs.length;
         }
       });
     } catch (error) {
@@ -518,60 +586,55 @@ export class SceneService implements OnModuleInit, OnModuleDestroy {
   //   );
   // }
 
-  // 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);
-  //       console.log('handleMoveSteam-onMoving', this.onMoving);
-  //       stream.marker = metaData.marker;
-  //       this.lastMoveStreamFrame.next(stream);
-  //       const next = this.frameCnt.value + 1;
-  //       this.currentMoveMaker = metaData.marker;
-  //       if (this.onMoving) {
-  //         this.frameCnt.next(next);
-  //       } else {
-  //         console.log('handleMoveSteam stop', next, this.currentMoveMaker);
-  //         this.cleanMoveSteam();
-  //         this.resumeStream(next);
-  //         return;
-  //       }
-  //       let src = stream.clipPath.split('?')[0];
-  //       // // 临时本地替换路经
-  //       src = src.replace('/0000000001/', '');
-  //       const clipPath = path.join(__dirname, `../ws/video/${src}`);
-
-  //       const streamData: StreamFrameType = {
-  //         frame: next,
-  //         clipPath: clipPath,
-  //         metaData: stream.metaData,
-  //         serverTime: this.mockserverTime,
-  //         DIR: 3,
-  //       };
-  //       await this.streamService.pushFrameToSteam(streamData);
-
-  //       if (this.lastMoveCnt == this.frameCnt.getValue()) {
-  //         const next = this.frameCnt.getValue() + 1;
-  //         console.log('last', next);
-  //         this.resumeStream(next);
-  //         this.cleanMoveSteam();
-  //         const lastFrame = this.lastMoveStreamFrame.getValue();
-  //         const userId = this.user_id;
-  //         const breakPointId = lastFrame.marker.split('T')[1];
-  //         const lastReply = JSON.parse(lastFrame.metaData);
-  //         this.moveService.updateUser(userId, breakPointId, lastReply);
-  //       }
-  //     },
-  //   );
-  // }
+  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);
+        console.log('handleMoveSteam-onMoving', this.onMoving.value);
+        stream.marker = metaData.marker;
+        this.lastMoveStreamFrame.next(stream);
+        const next = this.frameCnt.value + 1;
+        this.currentMoveMaker = metaData.marker;
+        if (this.onMoving) {
+          this.frameCnt.next(next);
+        } else {
+          console.log('handleMoveSteam stop', next, this.currentMoveMaker);
+          this.cleanMoveSteam();
+          this.resumeStream();
+          return;
+        }
+        const streamData: StreamFrameType = {
+          frame: next,
+          clipPath: stream.clipPath,
+          metaData: stream.metaData,
+          serverTime: this.mockserverTime,
+          DIR: 3,
+        };
+        await this.streamService.pushFrameToSteam(streamData);
+
+        if (this.lastMoveCnt == this.frameCnt.getValue()) {
+          const next = this.frameCnt.getValue() + 1;
+          console.log('last', next);
+          this.resumeStream();
+          this.cleanMoveSteam();
+          const lastFrame = this.lastMoveStreamFrame.getValue();
+          const userId = this.user_id;
+          const breakPointId = lastFrame.marker.split('T')[1];
+          const lastReply = JSON.parse(lastFrame.metaData);
+          this.moveService.updateUser(userId, breakPointId, lastReply);
+        }
+      },
+    );
+  }
 }

src/scene/walking-consumer.ts

@@ -9,6 +9,7 @@ import {
 } from '@nestjs/bull';
 import { Injectable } from '@nestjs/common';
 import { Job } from 'bull';
+import { MoveService } from 'src/move/move.service';
 import { SceneService } from './scene.service';
 import { StreamService } from './stream/stream.service';
 
@@ -18,6 +19,7 @@ export class WalkingConsumer {
   constructor(
     private streamService: StreamService,
     private sceneService: SceneService,
+    private moveService: MoveService,
   ) {}
   private isDone = true;
   private _checkerRotateDone: NodeJS.Timeout;
@@ -37,18 +39,24 @@ export class WalkingConsumer {
 
   @OnQueueDrained()
   onDrained(job: Job) {
-    console.log(`onDrained`);
+    console.log(`onDrained`, job);
   }
 
   @OnQueueCompleted()
   onQueueComplete(job: Job) {
-    console.log(`onQueueComplete-frame`, job.data.frame);
+    console.log(`walking-frame`, job.data.frame);
     this._checkerRotateDone = setTimeout(() => {
-      console.log('1秒后开流');
+      console.log('walking--1秒后开流');
       const next = job.data.frame + 1;
       this.sceneService.resumeStream();
       this.sceneService.onRotating.next(false);
       this.sceneService.frameCnt.next(next);
+
+      const { userId } = this.sceneService.getConfig();
+      const lastFrame = JSON.parse(job.data.metaData);
+      const breakPointId = lastFrame.marker.split('T')[1];
+      const lastReply = lastFrame;
+      this.moveService.updateUser(userId, breakPointId, lastReply);
     }, 1000);
   }