增加多个基础逻辑，上下线

log/.46f0af219249542fdc264b9b0460d72fee271bbf-audit.json

@@ -1,25 +0,0 @@
-{
-    "keep": {
-        "days": true,
-        "amount": 30
-    },
-    "auditLog": "D:\\codes\\daikan\\socket-server-v4\\log\\.46f0af219249542fdc264b9b0460d72fee271bbf-audit.json",
-    "files": [
-        {
-            "date": 1676369333465,
-            "name": "D:\\codes\\daikan\\socket-server-v4\\log\\2023-02-14.log",
-            "hash": "2ab3113bffdc301303be9f0789c1993a5dd376f54ab83789f1cafe128dbac5a8"
-        },
-        {
-            "date": 1676422429133,
-            "name": "D:\\codes\\daikan\\socket-server-v4\\log\\2023-02-15.log",
-            "hash": "b1410283ed4fb79aeeb9d673071592ae53cf16293856d13796ee9dd324fe8c08"
-        },
-        {
-            "date": 1676508231970,
-            "name": "D:\\codes\\daikan\\socket-server-v4\\log\\2023-02-16.log",
-            "hash": "eb40924a4d09ab101e29fbb5ff18f149f8ff001a793bb99c2c5ec52f63e6ed95"
-        }
-    ],
-    "hashType": "sha256"
-}

src/main.ts

@@ -26,7 +26,7 @@ async function bootstrap() {
             winston.format.timestamp(),
             winston.format.json(),
           ),
-          filename: join(__dirname, '..', 'log/%DATE%.log'),
+          filename: join(__dirname, '..', 'logs/%DATE%.log'),
           datePattern: 'YYYY-MM-DD',
           zippedArchive: true,
           maxSize: '20m',

src/room/actions/action.d.ts

@@ -18,7 +18,8 @@ type actionType =
   | 'answer-currentscene'
   | 'error'
   | 'sync-floor'
-  | 'room-valid-time';
+  | 'room-valid-time'
+  | 'users-kicked';
 
 interface DanmakuDataType {
   nickname: string;
@@ -38,3 +39,8 @@ interface TypingStateType {
   type: actionType['users-words'];
   userId: string;
 }
+
+interface KickStateType {
+  type: actionType['users-kicked'];
+  userId: string;
+}

src/room/actions/actions.service.ts

@@ -1,6 +1,7 @@
 import { InjectRedis } from '@liaoliaots/nestjs-redis';
 import { forwardRef, Inject, Injectable } from '@nestjs/common';
 import { Redis } from 'ioredis';
+import { Socket } from 'socket.io';
 import { RoomService } from '../room.service';
 import { UsersService } from '../users/users.service';
 @Injectable()
@@ -11,10 +12,12 @@ export class ActionsService {
     private roomService: RoomService,
     @Inject(forwardRef(() => UsersService))
     private userService: UsersService,
-  ) {}
+  ) { }
 
-  async handleAllAction(data: ActionsParams): Promise<void> {
-    // console.log('ActionsParams', data);
+  async handleAllAction(socket: Socket, data: ActionsParams): Promise<void> {
+    const isSocketLeader = () => {
+      return socket.data.user?.Role === 'leader';
+    };
     switch (data.type) {
       case 'danmumsg':
         const params = {} as DanmakuDataType;
@@ -25,28 +28,26 @@ export class ActionsService {
         this.handleDanmaku(params);
         break;
       case 'users-muted':
-        console.log('users-muted', data);
         const mutedParams = data as any as MutedStateType;
-        this.roomService._isLeader &&
-          (await this.handleMutedState(mutedParams));
+        isSocketLeader() && (await this.handleMutedState(mutedParams));
         break;
       case 'users-words':
         const typingParams = data as any as TypingStateType;
-        this.roomService._isLeader &&
-          (await this.handleTypingState(typingParams));
+        isSocketLeader() && (await this.handleTypingState(typingParams));
+        break;
+      case 'users-kicked':
+        const kickParams = data as any as KickStateType;
+        // console.log('kickParams', socket.data.user, kickParams);
+        isSocketLeader() && (await this.handleKickState(socket, kickParams));
         break;
       default:
-        this.answer(data);
         break;
     }
-    this.answer(data);
-  }
-  answer(data: any) {
-    // this.roomService.socketGateway._socket
-    //   .to(this.roomService._userInfo.RoomId)
-    //   .emit('action', data);
-    this.roomService.socketGateway._socket.broadcast.emit('action', data);
+    socket.broadcast.emit('action', data);
   }
+  // answer(data: any) {
+  //   this.roomService.socketGateway._socket.broadcast.emit('action', data);
+  // }
 
   async handleDanmaku(data: DanmakuDataType) {
     console.log('data', data);
@@ -82,4 +83,12 @@ export class ActionsService {
       await this.userService.updateUserTypingState(roomId, userId, wordsState);
     }
   }
+  async handleKickState(socket: Socket, data: KickStateType) {
+    if (data) {
+      const roomId = socket.data.user.RoomId;
+      const userId = data.userId;
+
+      await this.roomService.handleKickAction(socket, roomId, userId);
+    }
+  }
 }

src/room/room.service.ts

@@ -1,6 +1,7 @@
 import { InjectRedis } from '@liaoliaots/nestjs-redis';
 import { forwardRef, Inject, Injectable, Logger } from '@nestjs/common';
 import { Redis } from 'ioredis';
+import { Socket } from 'socket.io';
 
 import { SocketGateway } from 'src/socket/socket.gateway';
 import { ActionsService } from './actions/actions.service';
@@ -14,8 +15,9 @@ export class RoomService {
     @InjectRedis() private readonly redis: Redis,
     private readonly userService: UsersService,
     private readonly actionsService: ActionsService,
-  ) {}
+  ) { }
   public readonly logger = new Logger('user');
+  public _sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
   public _userInfo = {} as UserInfoType;
   private _roomConfig = {} as RoomConfigType;
   private isJoin = false;
@@ -35,13 +37,14 @@ export class RoomService {
    */
   initUserProfile(userInfo: UserInfoParams): void {
     this._userInfo = {
+      id: userInfo.id,
       RoomId: userInfo.roomId,
       Role: userInfo.role,
       UserId: userInfo.userId,
       Avatar: userInfo.avatar,
       Nickname:
         decodeURIComponent(userInfo.nickname) || `用户[${userInfo.userId}]`,
-      IsClient: userInfo.isClient,
+      IsClient: Boolean(userInfo.isClient),
       IsMuted: true,
       IsWords: true,
       Order: userInfo.role === 'leader' ? 0 : 1,
@@ -50,36 +53,29 @@ export class RoomService {
       IsOnline: true,
     };
   }
-  async handleUserOnline() {
-    this._userInfo.IsOnline = true;
-    if (this._userId && this._roomId) {
+  async handleUserOnline(socket: Socket) {
+    if (socket.data.user) {
+      socket.data.user.IsOnline = true;
       await this.userService.updateUserOnlineState(
-        this._roomId,
-        this._userId,
+        socket,
+        socket.data.user.RoomId,
+        socket.data.user.UserId,
         true,
       );
-      const roomUsers = await this.userService.getRoomUsers(this._roomId);
-      this.isJoin &&
-        this.socketGateway._socket.to(this._roomId).emit('action', {
-          type: 'rooms-status',
-          members: roomUsers,
-        });
     }
   }
 
-  async handleUserOffline() {
-    this._userInfo.IsOnline = false;
-    if (this._userId && this._roomId) {
+  async handleUserOffline(socket: Socket) {
+    await this._sleep(500);
+    if (socket.data.user) {
+      socket.data.user.IsOnline = false;
       await this.userService.updateUserOnlineState(
-        this._roomId,
-        this._userId,
+        socket,
+        socket.data.user.RoomId,
+        socket.data.user.UserId,
         false,
       );
-      const roomUsers = await this.userService.getRoomUsers(this._roomId);
-      this.socketGateway._socket.to(this._roomId).emit('action', {
-        type: 'rooms-status',
-        members: roomUsers,
-      });
+      await this.handleRoomStatusAction(socket);
     }
   }
 
@@ -88,8 +84,13 @@ export class RoomService {
    * @param userInfo
    */
 
-  async handleUserJoin(userInfo: UserInfoParams): Promise<void> {
+  async handleUserJoin(
+    socket: Socket,
+    userInfo: UserInfoParams,
+  ): Promise<void> {
     this.initUserProfile(userInfo);
+    socket.data.user = this._userInfo;
+    await this.handleUserOnline(socket);
     let blockJoin = false;
     if (this._roomId?.length && this._userId?.length) {
       //房主设置房间配置
@@ -106,7 +107,7 @@ export class RoomService {
           );
         } else {
           blockJoin = true;
-          this.socketGateway._socket.emit('action', {
+          socket.emit('action', {
             type: 'invalid-master',
             code: 303,
           });
@@ -124,7 +125,7 @@ export class RoomService {
         } else {
           const updated = await this.userService.updateUsers(this._userInfo);
           if (!updated) {
-            this.socketGateway._socket.emit('action', {
+            socket.emit('action', {
               type: 'error',
               code: 405,
             });
@@ -136,12 +137,12 @@ export class RoomService {
           }
         }
         const roomUsers = await this.userService.getRoomUsers(this._roomId);
-        this.socketGateway._socket.emit('join', {
-          user: this._userInfo,
+        socket.emit('join', {
+          user: socket.data.user,
           members: roomUsers,
         });
-        if (!this._userInfo.IsClient) {
-          this.socketGateway._socket.broadcast.to(this._roomId).emit('action', {
+        if (!socket.data.user.IsClient) {
+          socket.broadcast.to(socket.data.user.RoomId).emit('action', {
             type: 'user-join',
             user: this._userInfo,
             members: roomUsers,
@@ -149,12 +150,12 @@ export class RoomService {
         }
         this.isJoin = true;
         this.logger.log(
-          JSON.stringify(this._userInfo),
-          `join-user-${this._userInfo.Role}`,
+          JSON.stringify(socket.data.user),
+          `join-user-${socket.data.user.Role}`,
         );
       }
     } else {
-      this.socketGateway._socket.emit('action', { type: 'error', code: 403 });
+      socket.emit('action', { type: 'error', code: 403 });
       this.logger.warn(`403:not roomId and userId`, 'join-error');
     }
   }
@@ -162,50 +163,106 @@ export class RoomService {
    *  房间action大通道
    * @param message
    */
-  async handleUserAction(message: any) {
-    this.actionsService.handleAllAction(message);
+  async handleUserAction(socket: Socket, message: any) {
+    this.actionsService.handleAllAction(socket, message);
   }
 
   /**
    *  房间sync大通道
    * @param message
    */
-  async handleSyncAction(message: any) {
-    this.socketGateway._socket.broadcast.to(this._roomId).emit('sync', message);
+  async handleSyncAction(socket: Socket, message: any) {
+    socket.broadcast.to(this._roomId).emit('sync', message);
   }
 
   /**
    *  房间paint大通道
    * @param message
    */
-  async handlePaintAction(message: any) {
-    this.socketGateway._socket.broadcast
-      .to(this._roomId)
-      .emit('paint', message);
+  async handlePaintAction(socket: Socket, message: any) {
+    socket.broadcast.to(this._roomId).emit('paint', message);
   }
 
   /**
    *  房间T人
    * @param message
    */
-  async handleKickAction(userId: string) {
-    await this.userService.deleteRoomUser(this._roomId, userId);
-    this.socketGateway._socket.broadcast.to(this._roomId).emit('action', {
-      type: 'kick-user',
-      data: {
-        userId: userId,
-      },
+  async handleKickAction(socket: Socket, RoomId: string, userId: string) {
+    const delUser = await this.userService.getUsersByUserId(RoomId, userId);
+    this.logger.warn(
+      `RoomId: ${RoomId},userId:${userId} socketId:${delUser.id}`,
+      'kick-user',
+    );
+    const roomUsers = await this.userService.getRoomUsers(RoomId);
+    const filterRoomUser = roomUsers.filter((i) => i.UserId !== userId);
+    if (delUser) {
+      socket.data.isKick = true;
+      this.socketGateway.server.sockets.sockets.forEach((soc) => {
+        if (soc.id === delUser.id) {
+          soc.disconnect(true);
+        }
+      });
+    }
+    const res = await this.userService.deleteRoomUser(RoomId, userId);
+    // console.log('kick-user', delUser.id, RoomId, userId);
+    if (res) {
+      this.socketGateway.server.to(RoomId).emit('action', {
+        type: 'user-leave',
+        user: delUser,
+        members: filterRoomUser,
+      });
+    }
+  }
+
+  /**
+   *  房间全员状态更新
+   * @param message
+   */
+  async handleRoomStatusAction(socket: Socket) {
+    const roomId = socket.data.user.RoomId;
+    const roomUsers = await this.userService.getRoomUsers(roomId);
+    this.socketGateway.server.to(roomId).emit('action', {
+      type: 'rooms-status',
+      members: roomUsers,
     });
   }
   /**
+   *  房间主动退出逻辑
+   * @param message
+   */
+  async handleExitAction(socket: Socket, message: any) {
+    console.log('socket', socket.data.user);
+    const roomId = message?.roomId || socket.data.user?.RoomId;
+    const userId = message?.userId || socket.data.user?.UserId;
+    if (roomId && userId) {
+      const delUser = await this.userService.getUsersByUserId(roomId, userId);
+      this.logger.warn(
+        `RoomId: ${roomId},userId:${userId} socketId:${delUser.id}`,
+        'room-exit',
+      );
+      const roomUsers = await this.userService.getRoomUsers(roomId);
+      const filterRoomUser = roomUsers.filter((i) => i.UserId !== userId);
+      const res = await this.userService.deleteRoomUser(roomId, userId);
+      console.log('user-exit', delUser.id, roomId, userId);
+      if (res) {
+        this.socketGateway.server.to(roomId).emit('action', {
+          type: 'user-leave',
+          user: delUser,
+          members: filterRoomUser,
+        });
+        socket.disconnect(true);
+      }
+    }
+  }
+  /**
    * 解散房间
    */
 
-  async handleRoomDismiss(): Promise<void> {
-    await this.userService.delRoom(this._roomId);
-    await this.userService.delRoomConfig(this._roomId);
-    this.socketGateway._socket
-      .to(this._roomId)
+  async handleRoomDismiss(RoomId: string): Promise<void> {
+    await this.userService.delRoom(RoomId);
+    await this.userService.delRoomConfig(RoomId);
+    this.socketGateway.server
+      .to(RoomId)
       .emit('action', { type: 'leader-dismiss' });
   }
 }

src/room/user.d.ts

@@ -1,4 +1,5 @@
 interface UserInfoType {
+  id?: string; //socketId
   UserId: string;
   RoomId: string;
   Role: string;
@@ -14,6 +15,7 @@ interface UserInfoType {
 }
 
 interface UserInfoParams {
+  id?: string; //socket id
   userId: string;
   roomId: string;
   role: string;

src/room/users/users.service.ts

@@ -1,10 +1,16 @@
 import { InjectRedis } from '@liaoliaots/nestjs-redis';
-import { Injectable } from '@nestjs/common';
+import { forwardRef, Inject, Injectable } from '@nestjs/common';
 import { Redis } from 'ioredis';
+import { Socket } from 'socket.io';
+import { RoomService } from '../room.service';
 
 @Injectable()
 export class UsersService {
-  constructor(@InjectRedis() private readonly redis: Redis) {}
+  constructor(
+    @InjectRedis() private readonly redis: Redis,
+    @Inject(forwardRef(() => RoomService))
+    private roomService: RoomService,
+  ) { }
 
   async isUserInRooms(RoomId: string, UserId: string): Promise<boolean> {
     const res = await this.redis.hexists(
@@ -14,13 +20,28 @@ export class UsersService {
 
     return res > 0 ? true : false;
   }
-  insertUser(useInfo: UserInfoType): Promise<any> {
-    return this.redis.hset(
+  async insertUser(useInfo: UserInfoType): Promise<boolean> {
+    const insert = await this.redis.hset(
       `kankan:socket:rooms:${useInfo.RoomId}`,
       useInfo.UserId,
       JSON.stringify(useInfo),
     );
+    return Promise.resolve(Boolean(insert));
   }
+  async getUsersByUserId(
+    RoomId: string,
+    UserId: string,
+  ): Promise<UserInfoType> {
+    const user = await this.redis.hget(`kankan:socket:rooms:${RoomId}`, UserId);
+    return JSON.parse(user) as unknown as UserInfoType;
+  }
+
+  async isUserLogin(RoomId: string, UserId: string): Promise<boolean> {
+    const user = await this.redis.hget(`kankan:socket:rooms:${RoomId}`, UserId);
+    const userObj = JSON.parse(user) as unknown as UserInfoType;
+    return userObj.IsOnline;
+  }
+
   async updateUsers(useInfo: UserInfoType): Promise<boolean> {
     const user = await this.redis.hget(
       `kankan:socket:rooms:${useInfo.RoomId}`,
@@ -45,16 +66,24 @@ export class UsersService {
     }
   }
 
-  async deleteRoomUser(RoomId: string, UserId: string) {
-    return this.redis.hdel(`kankan:socket:rooms:${RoomId}`, UserId);
+  async deleteRoomUser(RoomId: string, UserId: string): Promise<boolean> {
+    const isDel = await this.redis.hdel(
+      `kankan:socket:rooms:${RoomId}`,
+      UserId,
+    );
+    return Promise.resolve(Boolean(isDel));
   }
 
   async getRoomUsers(RoomId: string): Promise<UserInfoType[]> {
     const res = await this.redis.hvals(`kankan:socket:rooms:${RoomId}`);
     const allUsers = Array.from(res)
-      .map((i) => JSON.parse(i))
+      .map((i) => {
+        const data = JSON.parse(i);
+        delete data['id'];
+        return data;
+      })
       .sort((a, b) => parseInt(a.Order) - parseInt(b.Order));
-    return allUsers as any as UserInfoType[];
+    return Promise.resolve(allUsers as any as UserInfoType[]);
   }
   delRoom(RoomId: string): Promise<any> {
     return this.redis.del(`kankan:socket:rooms:${RoomId}`);
@@ -87,17 +116,22 @@ export class UsersService {
   }
 
   async updateUserOnlineState(
+    socket: Socket,
     RoomId: string,
     UserId: string,
-    status = true,
+    status: boolean,
   ): Promise<void> {
     const user = await this.redis.hget(`kankan:socket:rooms:${RoomId}`, UserId);
     if (user) {
       const userObj: UserInfoType = JSON.parse(user);
       const updateObj: UserInfoType = Object.assign({}, userObj, {
+        id: socket.id,
         IsOnline: status,
+        IsMuted: true, //强制上线或下线时静音
       });
-      this.redis.hset(
+      // console.warn('updateUserOnlineState', RoomId, UserId, status);
+      // console.log('updateObj', JSON.stringify(updateObj));
+      await this.redis.hset(
         `kankan:socket:rooms:${RoomId}`,
         UserId,
         JSON.stringify(updateObj),

src/socket/socket.gateway.ts

@@ -14,6 +14,7 @@ import Redis from 'ioredis';
 import { instrument } from '@socket.io/admin-ui';
 import * as bcrypt from 'bcrypt';
 import { RoomService } from '../room/room.service';
+import { Logger } from '@nestjs/common';
 
 console.log('SOCKET_PATH-0', process.env.SOCKET_PATH);
 @WebSocketGateway(Number(process.env.SOCKET_PORT), {
@@ -30,17 +31,23 @@ export class SocketGateway
     @InjectRedis() private readonly redis: Redis,
     private readonly roomService: RoomService,
   ) { }
-
+  public readonly logger = new Logger('socketGateway');
   @WebSocketServer() server: Server;
-  public _socket: Socket;
-  async handleConnection(client: Socket, ...args: any[]) {
-    // console.log('handleConnection', client, args);
-    this._socket = client;
-    await this.roomService.handleUserOnline();
+  public _loginLimit = new Map<string, string>();
+
+  async handleConnection(@ConnectedSocket() socket: Socket, ...args: any[]) {
   }
-  async handleDisconnect(client: Socket) {
-    // this._socket = client;
-    await this.roomService.handleUserOffline();
+  async handleDisconnect(@ConnectedSocket() socket: Socket) {
+    const deviceId = socket.data.deviceId;
+    if (deviceId) {
+      const did = this._loginLimit.get(deviceId);
+      if (did === socket.id) {
+        this._loginLimit.delete(deviceId);
+      }
+    }
+    await this.roomService.handleUserOffline(socket);
+    socket.removeAllListeners();
+    socket.disconnect();
     // console.log('handleDisconnect', client);
   }
   afterInit(server: any) {
@@ -59,24 +66,61 @@ export class SocketGateway
     @MessageBody() message: UserInfoParams,
     @ConnectedSocket() socket: Socket,
   ): Promise<void> {
-    socket.join(message.roomId);
-    await this.roomService.handleUserJoin(message);
+    message.id = socket.id;
+    // (socket as any).userId = message.userId;
+    // (socket as any).roomId = message.roomId;
+    socket.data.user = message;
+    const isRepeat = this.handleRepeatJoin(socket, message);
+    if (!isRepeat) {
+      socket.join(message.roomId);
+      await this.roomService.handleUserJoin(socket, message);
+    }
+  }
+  handleRepeatJoin(socket: Socket, message: UserInfoParams): boolean {
+    const from = message.isClient ? 0 : 1;
+    const flag = `${message.userId}-${from}`;
+    socket.data.deviceId = flag;
+    if (!this._loginLimit.has(flag)) {
+      this._loginLimit.set(flag, socket.id);
+      return false;
+    } else {
+      socket.emit('repeat-login', { isRepeat: true });
+      this.logger.warn(`306:${message.userId}`, 'repeat-login');
+      return true;
+    }
   }
 
   // 订阅action通道
   @SubscribeMessage('action')
-  async handleActionMessage(@MessageBody() message: any): Promise<void> {
-    await this.roomService.handleUserAction(message);
+  async handleActionMessage(
+    @ConnectedSocket() socket: Socket,
+    @MessageBody() message: any,
+  ): Promise<void> {
+    await this.roomService.handleUserAction(socket, message);
   }
 
   // 订阅sync通道
   @SubscribeMessage('sync')
-  async handleSyncMessage(@MessageBody() message: any): Promise<void> {
-    await this.roomService.handleSyncAction(message);
+  async handleSyncMessage(
+    @ConnectedSocket() socket: Socket,
+    @MessageBody() message: any,
+  ): Promise<void> {
+    await this.roomService.handleSyncAction(socket, message);
   }
   // 订阅paint通道
   @SubscribeMessage('paint')
-  async handlePaintessage(@MessageBody() message: any): Promise<void> {
-    await this.roomService.handlePaintAction(message);
+  async handlePaintessage(
+    @ConnectedSocket() socket: Socket,
+    @MessageBody() message: any,
+  ): Promise<void> {
+    await this.roomService.handlePaintAction(socket, message);
+  }
+  // 订阅主动退出exit通道
+  @SubscribeMessage('exit')
+  async handleExitessage(
+    @ConnectedSocket() socket: Socket,
+    @MessageBody() message: any,
+  ): Promise<void> {
+    await this.roomService.handleExitAction(socket, message);
   }
 }