Adapting admin connection to pusher.

Done: onUserLeave / onUserEnter
To be done: ban / unban

pusher/src/Controller/IoSocketController.ts

@@ -24,6 +24,7 @@ import {emitInBatch} from "../Services/IoSocketHelpers";
 import {clientEventsEmitter} from "../Services/ClientEventsEmitter";
 import {ADMIN_API_TOKEN, ADMIN_API_URL, SOCKET_IDLE_TIMER} from "../Enum/EnvironmentVariable";
 import {Zone} from "_Model/Zone";
+import {ExAdminSocketInterface} from "_Model/Websocket/ExAdminSocketInterface";
 
 export class IoSocketController {
     private nextUserId: number = 1;
@@ -46,7 +47,12 @@ export class IoSocketController {
                     res.writeStatus("401 Unauthorized").end('Incorrect token');
                     return;
                 }
-                const roomId = query.roomId as string;
+                const roomId = query.roomId;
+                if (typeof roomId !== 'string') {
+                    console.error('Received')
+                    res.writeStatus("400 Bad Request").end('Missing room id');
+                    return;
+                }
 
                 res.upgrade(
                     {roomId},
@@ -55,7 +61,12 @@ export class IoSocketController {
             },
             open: (ws) => {
                 console.log('Admin socket connect for room: '+ws.roomId);
-                ws.send('Data:'+JSON.stringify(socketManager.getAdminSocketDataFor(ws.roomId as string)));
+                const roomId = ws.roomId;
+                ws.disconnecting = false;
+
+                socketManager.handleAdminRoom(ws as ExAdminSocketInterface, ws.roomId as string);
+
+                /*ws.send('Data:'+JSON.stringify(socketManager.getAdminSocketDataFor(ws.roomId as string)));
                 ws.clientJoinCallback = (clientUUid: string, roomId: string) => {
                     const wsroomId = ws.roomId as string;
                     if(wsroomId === roomId) {
@@ -69,7 +80,7 @@ export class IoSocketController {
                     }
                 };
                 clientEventsEmitter.registerToClientJoin(ws.clientJoinCallback);
-                clientEventsEmitter.registerToClientLeave(ws.clientLeaveCallback);
+                clientEventsEmitter.registerToClientLeave(ws.clientLeaveCallback);*/
             },
             message: (ws, arrayBuffer, isBinary): void => {
                 try {
@@ -101,9 +112,15 @@ export class IoSocketController {
                 }
             },
             close: (ws, code, message) => {
-                //todo make sure this code unregister the right listeners
-                clientEventsEmitter.unregisterFromClientJoin(ws.clientJoinCallback);
-                clientEventsEmitter.unregisterFromClientLeave(ws.clientLeaveCallback);
+                const Client = (ws as ExAdminSocketInterface);
+                try {
+                    Client.disconnecting = true;
+                    //leave room
+                    socketManager.leaveAdminRoom(Client);
+                } catch (e) {
+                    console.error('An error occurred on admin "disconnect"');
+                    console.error(e);
+                }
             }
         })
     }

pusher/src/Model/Websocket/ExAdminSocketInterface.ts

@@ -0,0 +1,21 @@
+import {PointInterface} from "./PointInterface";
+import {Identificable} from "./Identificable";
+import {ViewportInterface} from "_Model/Websocket/ViewportMessage";
+import {
+    AdminPusherToBackMessage,
+    BatchMessage,
+    PusherToBackMessage, ServerToAdminClientMessage,
+    ServerToClientMessage,
+    SubMessage
+} from "../../Messages/generated/messages_pb";
+import {WebSocket} from "uWebSockets.js"
+import {CharacterTexture} from "../../Services/AdminApi";
+import {ClientDuplexStream} from "grpc";
+import {Zone} from "_Model/Zone";
+
+export type AdminConnection = ClientDuplexStream<AdminPusherToBackMessage, ServerToAdminClientMessage>;
+
+export interface ExAdminSocketInterface extends WebSocket {
+    adminConnection: AdminConnection,
+    disconnecting: boolean,
+}

pusher/src/Services/SocketManager.ts

@@ -14,16 +14,23 @@ import {
     SilentMessage,
     SubMessage,
     ReportPlayerMessage,
-    UserJoinedMessage, UserLeftMessage,
+    UserJoinedMessage,
+    UserLeftMessage,
     UserMovedMessage,
     UserMovesMessage,
-    ViewportMessage, WebRtcDisconnectMessage,
+    ViewportMessage,
+    WebRtcDisconnectMessage,
     WebRtcSignalToClientMessage,
     WebRtcSignalToServerMessage,
     WebRtcStartMessage,
     QueryJitsiJwtMessage,
     SendJitsiJwtMessage,
-    SendUserMessage, JoinRoomMessage, CharacterLayerMessage, PusherToBackMessage
+    SendUserMessage,
+    JoinRoomMessage,
+    CharacterLayerMessage,
+    PusherToBackMessage,
+    AdminPusherToBackMessage,
+    ServerToAdminClientMessage
 } from "../Messages/generated/messages_pb";
 import {PointInterface} from "../Model/Websocket/PointInterface";
 import {ProtobufUtils} from "../Model/Websocket/ProtobufUtils";
@@ -42,6 +49,7 @@ import {apiClientRepository} from "./ApiClientRepository";
 import {ServiceError} from "grpc";
 import {GroupDescriptor, UserDescriptor, ZoneEventListener} from "_Model/Zone";
 import Debug from "debug";
+import {ExAdminSocketInterface} from "_Model/Websocket/ExAdminSocketInterface";
 
 const debug = Debug('socket');
 
@@ -70,6 +78,54 @@ export class SocketManager implements ZoneEventListener {
         });
     }
 
+    async handleAdminRoom(client: ExAdminSocketInterface, roomId: string): Promise<void> {
+        console.log('Calling adminRoom')
+        const apiClient = await apiClientRepository.getClient(roomId);
+        const adminRoomStream = apiClient.adminRoom();
+        client.adminConnection = adminRoomStream;
+
+        adminRoomStream.on('data', (message: ServerToAdminClientMessage) => {
+            if (message.hasUseruuidjoinedroom()) {
+                const userUuid = message.getUseruuidjoinedroom();
+
+                if (!client.disconnecting) {
+                    client.send('MemberJoin:'+userUuid+';'+roomId);
+                }
+            } else if (message.hasUseruuidleftroom()) {
+                const userUuid = message.getUseruuidleftroom();
+
+                if (!client.disconnecting) {
+                    client.send('MemberLeave:'+userUuid+';'+roomId);
+                }
+            } else {
+                throw new Error('Unexpected admin message');
+            }
+        }).on('end', () => {
+            console.warn('Admin connection lost to back server');
+            // Let's close the front connection if the back connection is closed. This way, we can retry connecting from the start.
+            if (!client.disconnecting) {
+                this.closeWebsocketConnection(client, 1011, 'Connection lost to back server');
+            }
+            console.log('A user left');
+        }).on('error', (err: Error) => {
+            console.error('Error in connection to back server:', err);
+            if (!client.disconnecting) {
+                this.closeWebsocketConnection(client, 1011, 'Error while connecting to back server');
+            }
+        });
+
+        const message = new AdminPusherToBackMessage();
+        message.setSubscribetoroom(roomId);
+
+        adminRoomStream.write(message);
+    }
+
+    leaveAdminRoom(socket : ExAdminSocketInterface) {
+        if (socket.adminConnection) {
+            socket.adminConnection.end();
+        }
+    }
+
     getAdminSocketDataFor(roomId:string): AdminSocketData {
         throw new Error('Not reimplemented yet');
         /*const data:AdminSocketData = {
@@ -205,7 +261,7 @@ export class SocketManager implements ZoneEventListener {
         }
     }
 
-    private closeWebsocketConnection(client: ExSocketInterface, code: number, reason: string) {
+    private closeWebsocketConnection(client: ExSocketInterface|ExAdminSocketInterface, code: number, reason: string) {
         client.disconnecting = true;
         //this.leaveRoom(client);
         //client.close();