/* * Copyright (c) 2022. * Author Peter Placzek (tada5hi) * For the full copyright and license information, * view the LICENSE file that was distributed with this source code. */ import {NamespaceInterface, ServerInterface, SocketInterface} from "../../../../modules/socket"; import {useChatRoomStatus} from "./user-status"; import {RoomEvent, useChatRoomEvent} from "./event"; import {useChatRoomUserCache, UserEventCache} from "./user-cache"; import {RedisAdapter} from "socket.io-redis"; import {ChatRoom, ChatRoomType} from "@chamesu/common/domains/chat/room"; import {ChatMember} from "@chamesu/common/domains/chat/member"; import {AppContext} from "../../../index"; import {RecordStatus} from "../../../../modules/socket/status"; export interface RoomContext { socketNamespace: NamespaceInterface, status: RecordStatus, event: RoomEvent, cache: UserEventCache } export class Room { public readonly ctx : RoomContext; protected loadRoomCallback : (roomId: string) => Promise | undefined; protected loadMemberCallback: (roomId: string, userId: number) => Promise | undefined; protected rooms: Map; protected members: WeakMap<{roomId: string, userId: number}, ChatMember>; //-------------------------------------------------------------------- constructor(ctx: RoomContext) { this.ctx = ctx; this.rooms = new Map(); this.members = new WeakMap<{roomId: string; userId: number}, ChatMember>(); this.subscribeCache(); } async cacheHandler(key: string) { const idStr : string = key.substr(this.ctx.cache.getKey().length + 1); const parts : string[] = idStr.split(':'); const roomId: string = parts[0]; const userId: number = parseInt(parts[1]); if(Number.isNaN(userId)) return; await this.ctx.status.setOffline(userId, roomId); await this.ctx.event.emitUserOffline(roomId, userId); } subscribeCache() { this.ctx.cache.on('expired', this.cacheHandler.bind(this)); } unsubscribeCache() { this.ctx.cache.on('expired', this.cacheHandler.bind(this)); } //-------------------------------------------------------------------- public setLoadRoomCallback(roomCallback: (roomId: string) => Promise) { this.loadRoomCallback = roomCallback; } public setLoadMemberCallback(memberCallback: (roomId: string, userId: number) => Promise) { this.loadMemberCallback = memberCallback; } //-------------------------------------------------------------------- async loadRoom(roomId: string) { if(this.rooms.has(roomId)) { return this.rooms.get(roomId); } if(typeof this.loadRoomCallback === 'undefined') { throw new Error('Es könenn keine Informationen über den Raum geladen werden...'); } const room = await this.loadRoomCallback(roomId); this.rooms.set(roomId, room); return room; } async loadMember(roomId: string, userId: number) { const key = {roomId, userId}; if(this.members.has(key)) { //return this.members.get(key); } if(typeof this.loadMemberCallback === 'undefined') { throw new Error('Es können keine Informationen über die Relation zwischen Raum und User geladen werden...'); } const member = await this.loadMemberCallback(roomId, userId); if(typeof member === 'undefined') { throw new Error('Es wurde keine Relation gefunden...'); } this.members.set(key, member); return member; } async isRoomUsable(socket: SocketInterface, roomId: string) : Promise { if(typeof socket.user === 'undefined') { return false; } const room = await this.loadRoom(roomId); if(typeof room === 'undefined') { throw new Error('Der Raum konnte nicht geladen werden...'); } if(room.type === ChatRoomType.PRIVATE) { try { await this.loadMember(roomId, socket.user.id); return true; } catch (e) { console.log(e); return false; } } return true; } //-------------------------------------------------------------------- /** * Join a chat room. * * @param socket * @param roomId */ async join(socket: SocketInterface, roomId: string) { if(!await this.isRoomUsable(socket, roomId)) { throw new Error('Der Raum ist nicht zugänglich...'); } const userId = socket.user.id; const socketId = socket.id; const isTimedOut = await this.ctx.cache.isExpired({roomId, userId}); if(!isTimedOut) { await this.ctx.cache.drop({roomId, userId}); } const sockets : Set = await this.getUserSockets(roomId, userId, socketId); if(sockets.size > 0) { await this.disconnectSockets(roomId, userId, sockets); } if(sockets.size === 0 && isTimedOut) { console.log('ChatRoom: #'+roomId+' User Status: user #' + socket.user.name + ' established a new connection.'); await this.ctx.status.setOnline(userId, roomId); await this.ctx.event.emitUserOnline(roomId, userId); } else { console.log('ChatRoom: #'+roomId+' User Status: user #' + socket.user.name + ' reconnected and established a connection.'); } socket .join(this.ctx.event.getFullSocketRoomKey(roomId)); socket .join(this.ctx.event.getFullSocketRoomUserKey(roomId, userId)); this.ctx.event.subscribeAsSocket(socket, roomId, 'messages'); this.ctx.event.subscribeAsSocket(socket, roomId, 'users'); } /** * Leave a chat room. * * @param socket * @param roomId */ async leave(socket: SocketInterface, roomId: string) { if(typeof socket.user === 'undefined') return; const userId = socket.user.id; const socketId = socket.id; const sockets : Set = await this.getUserSockets(roomId, userId, socketId); if(sockets.size === 0) { await this.ctx.cache.set({roomId, userId}); } console.log('ChatRoom: #'+roomId+' User Status: user #' + socket.user.name+' dropped out. Online status will expire soon...'); socket .leave(this.ctx.event.getFullSocketRoomKey(roomId)); socket .leave(this.ctx.event.getFullSocketRoomUserKey(roomId, userId)); this.ctx.event.unsubscribeAsSocket(socket, roomId, 'messages'); this.ctx.event.unsubscribeAsSocket(socket, roomId, 'users'); } //-------------------------------------------------------------------- /** * Get user socket ids which are connected * with the specific room. * * @param roomId * @param userId * @param expectSocketId */ protected async getUserSockets(roomId: string, userId: number, expectSocketId?: string) : Promise> { const roomUserKey = this.ctx.event.getFullSocketRoomUserKey(roomId, userId); const clients = await this.ctx .socketNamespace .in(roomUserKey) .allSockets(); if(typeof expectSocketId !== 'undefined') { if(clients.has(expectSocketId)) { clients.delete(expectSocketId); } } return clients; } /** * Disconnect specific sockets from chat room. * * @param roomId * @param userId * @param socketIds */ protected async disconnectSockets(roomId: string, userId: number, socketIds: Set) { const adapter : RedisAdapter = this.ctx.socketNamespace.adapter as unknown as RedisAdapter; const ids : string[] = Array.from(socketIds); for(let i=0; i { const key: string = this.ctx.status.options.redisPrefix+'.status.'+this.ctx.status.options.redisSuffix+':*'; const keys = await this.ctx.status.redisDatabase.keys(key); if(keys.length === 0) { return; } for(let i=0; i console.info('Chat Room: #'+id+' User Status: Closed #' + online.data[i].id + ' connections.')); } } } } } let room : Room | undefined; export function useChatRoom( ctx?: AppContext, roomCallback?: (roomId: string) => Promise, memberCallback?: (roomId: string, userId: number) => Promise ) : Room { if(typeof room !== 'undefined') { return room; } if(typeof ctx === 'undefined') { throw new Error('Context was not injected in room handler...'); } const namespaceName = '/chat'; const namespace = ctx.socketServer.of(namespaceName); const roomStatus = useChatRoomStatus(ctx.config.redisDatabase); const roomEvent = useChatRoomEvent(namespaceName); const roomUserCache = useChatRoomUserCache(ctx.config.redisDatabase); room = new Room( { socketNamespace: namespace, event: roomEvent, status: roomStatus, cache: roomUserCache } ); if(typeof roomCallback === 'function') { room.setLoadRoomCallback(roomCallback); } if(typeof memberCallback === 'function') { room.setLoadMemberCallback(memberCallback); } return room; }