import { IPCManager , ICreateRoomData , IJoinRoomData , ILeaveRoomData , ICloseRoomData , IUpdateRoomData , IRoomUpdate , IRoomObject , RoomManager , logger , } from "./internal" ; const log = logger ( "balancer" , 34 ) ; interface IProcesses { [ pid ] : { rooms : number ; clients : number ; } ; } export class LoadBalancer { constructor ( public rooms , public ipcm ) { ipcm . onRequest ( "createRoom" , ( data ) => this . rooms . createRoom ( data ) ) ; ipcm . onRequest ( "joinRoom" , ( data ) => this . rooms . joinRoom ( data ) ) ; ipcm . onRequest ( "updateRoom" , ( data ) => this . rooms . updateRoom ( data ) ) ; ipcm . onRequest ( "leaveRoom" , ( data ) => this . rooms . leaveRoom ( data ) ) ; ipcm . onRequest ( "closeRoom" , ( data ) => this . rooms . closeRoom ( data ) ) ; } public async findProcessForRoom ( ) { const pids = await this . ipcm . pids ( ) ; log . debug ( `Find process for room:` ) ; const roomsCache = await this . rooms . cache . findMany ( ) ; const processes = { } ; roomsCache . forEach ( ( room ) => { if ( pids . includes ( String ( room . pid ) ) ) { const { rooms = 0 , clients = 0 } = processes [ room . pid ] || { } ; processes [ room . pid ] = { rooms : rooms + 1 , clients : clients + room . clients . length } ; } else { this . rooms . cache . remove ( room . key ) ; } } ) ; log . debug ( `Current process load:` , processes ) ; let minLoad = Infinity ; let roomProcessId = this . ipcm . processId ; log . debug ( `All cluster pids` , pids ) ; for ( const pid of pids ) { const { rooms = 0 , clients = 0 } = processes [ pid ] || { } ; if ( rooms * clients >= minLoad ) { continue ; } roomProcessId = pid ; minLoad = rooms * clients ; } return roomProcessId ; } public async createRoom ( sessionId , name , options ? ) { const pid = await this . findProcessForRoom ( ) ; if ( pid === this . ipcm . processId ) { return this . rooms . createRoom ( { sessionId , name , options } ) ; } return this . ipcm . requestProcess < ICreateRoomData , IRoomObject > ( pid , "createRoom" , { sessionId , name , options } ) ; } public async makeRequest ( method , roomId , data ) { const room = await this . rooms . cache . get ( roomId ) ; if ( ! room ) { throw new Error ( `Cannot ${ method } with id ${ roomId } - room not found.` ) ; } if ( room . pid === this . ipcm . processId ) { switch ( method ) { case "joinRoom" : return this . rooms . joinRoom ( data ) ; case "updateRoom" : return this . rooms . updateRoom ( data ) ; case "leaveRoom" : return this . rooms . leaveRoom ( data ) ; case "closeRoom" : return this . rooms . closeRoom ( data ) ; } } return this . ipcm . requestProcess ( room . pid , method , data ) ; } public async joinRoom ( sessionId , roomId , options ? ) { return this . makeRequest ( "joinRoom" , roomId , { sessionId , roomId , options } ) ; } public async updateRoom ( sessionId , roomId , update ) { return this . makeRequest ( "updateRoom" , roomId , { sessionId , roomId , update } ) ; } public async leaveRoom ( sessionId , roomId ) { return this . makeRequest ( "leaveRoom" , roomId , { sessionId , roomId } ) ; } public async closeRoom ( sessionId , roomId ) { return this . makeRequest ( "closeRoom" , roomId , { sessionId , roomId } ) ; } }