Merge pull request #234 from TUMFARSynchrony/ed-websocket-lobby-redirect

Author: NCEghtebas

backend/hub/hub.py

@@ -5,6 +5,8 @@
 import asyncio
 import logging
 
+from jsonpickle import json
+
 from custom_types.message import MessageDict
 from session.data.participant.participant_summary import ParticipantSummaryDict
 from connection.messages.rtc_ice_candidate_dict import RTCIceCandidateDict
@@ -122,6 +124,13 @@ def remove_experimenter(self, experimenter: Experimenter):
         """
         self.experimenters.remove(experimenter)
 
+    async def notify_participants_experimenter_joined(self, message):
+        """Handle the event and broadcast a WebSocket message."""
+        self._logger.info("Notifying participants experimenter joined")
+        await self.server.broadcast_message(json.dumps(message))
+        self._logger.info("Notified participants")
+
+
     async def handle_offer(
         self,
         offer: RTCSessionDescription,

backend/server/server.py

@@ -4,7 +4,7 @@
 import aiohttp_cors
 import json
 from typing import Any, Callable, Coroutine, Literal
-from aiohttp import web
+from aiohttp import web, WSMsgType
 from datetime import datetime
 from aiortc import RTCSessionDescription
 from ssl import SSLContext
@@ -64,12 +64,14 @@ def __init__(
         self._hub_handle_add_ice_candidate = hub_handle_add_ice_candidate
         self._config = config
 
+        self.websockets = set()
         self._app = web.Application()
         self._app.on_shutdown.append(self._shutdown)
         routes = [
             self._app.router.add_get("/hello-world", self.get_hello_world),
             self._app.router.add_post("/offer", self.handle_offer),
             self._app.router.add_post("/addIceCandidate", self.handle_add_ice_candidate),
+            self._app.router.add_get('/ws', self.websocket_handler)
         ]
 
         # Serve frontend build
@@ -146,6 +148,28 @@ async def start(self):
         )
         await site.start()
 
+    async def websocket_handler(self, request):
+        """Handle incoming WebSocket connections."""
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        self.websockets.add(ws)
+
+        async for msg in ws:
+            if msg.type == WSMsgType.TEXT:
+                # Handle incoming WebSocket messages if needed
+                pass
+            elif msg.type == WSMsgType.ERROR:
+                print('WebSocket connection closed with exception %s' % ws.exception())
+
+        # Cleanup on disconnect
+        self.websockets.remove(ws)
+        return ws
+
+    async def broadcast_message(self, message):
+        """Send a message to all connected clients."""
+        for ws in self.websockets:
+            await ws.send_str(message)
+
     async def _shutdown(self, app: web.Application):
         """TODO document"""
         pass

backend/users/experimenter.py

@@ -320,6 +320,9 @@ async def _handle_create_experiment(self, data: Any) -> MessageDict:
 
         self._experiment = await self._hub.create_experiment(data["session_id"])
         self._experiment.add_experimenter(self)
+        self._logger.info(f"sending ws message. {str(self)}")
+
+        await self._hub.notify_participants_experimenter_joined({"type": "EXPERIMENTER_JOINED"})
 
         # Subscribe to participants in experiment
         await self._subscribe_to_participants_streams()
@@ -386,6 +389,9 @@ async def _handle_join_experiment(self, data: Any) -> MessageDict:
 
         self._experiment = experiment
         self._experiment.add_experimenter(self)
+        self._logger.info(f"sending ws message. {str(self)}")
+
+        await self._hub.notify_participants_experimenter_joined({"type": "EXPERIMENTER_JOINED"})
 
         # Subscribe to participants in experiment
         await self._subscribe_to_participants_streams()

frontend/src/pages/Lobby/Lobby.js

@@ -13,6 +13,7 @@ import { InstructionsTab } from "../../components/molecules/InstructionsTab/Inst
 import "./Lobby.css";
 import { ParticipantChatTab } from "../../components/molecules/ChatTab/ParticipantChatTab";
 import { ChatGptTab } from "../../components/molecules/ChatGptTab/ChatGptTab";
+import { BACKEND } from "../../utils/constants";
 import VideoCanvas from "../../components/organisms/VideoCanvas/VideoCanvas";
 import { ActionButton } from "../../components/atoms/Button";
 
@@ -63,6 +64,47 @@ function Lobby({ localStream, connection, onGetSession, onChat }) {
     }
   }, [localStream, participantStream]);
 
+  useEffect(() => {
+    const wsUrl = `${BACKEND.replace(/^http/, "ws")}/ws`;
+    console.log(`Connecting to WebSocket at ${wsUrl}`);
+
+    console.log(`Attempting to connect to WebSocket at ${wsUrl}`);
+
+    const ws = new WebSocket(wsUrl);
+
+    ws.onopen = () => {
+      console.log("WebSocket connection established");
+    };
+
+    ws.onmessage = (event) => {
+      const message = JSON.parse(event.data);
+      if (message.type === "EXPERIMENTER_JOINED") {
+        // Handle the experimenter joined message appropriately
+        console.log("Experimenter connected");
+        window.location.reload();
+      }
+    };
+
+    ws.onerror = (error) => {
+      console.error("WebSocket error observed:", error);
+    };
+
+    ws.onclose = (event) => {
+      console.log("WebSocket connection closed:", event.reason);
+    };
+
+    const handleUnload = () => {
+      ws.close();
+    };
+
+    window.addEventListener("beforeunload", handleUnload);
+
+    return () => {
+      window.removeEventListener("beforeunload", handleUnload);
+      ws.close();
+    };
+  }, []);
+
   const streamChangeHandler = async () => {
     console.log("%cRemote Stream Change Handler", "color:blue");
   };