🎨 Removes stringified data field in socketio message (#5335)

pcrespov · ignapas · web-flow · commit 226c183f9a47 · 2024-02-22T12:53:17.000Z
Co-authored-by: Ignacio Pascual &lt;4764217+ignapas@users.noreply.github.com&gt;
diff --git a/packages/pytest-simcore/src/pytest_simcore/playwright_utils.py b/packages/pytest-simcore/src/pytest_simcore/playwright_utils.py
@@ -53,7 +53,7 @@ class SocketIOEvent:
 
 def decode_socketio_42_message(message: str) -> SocketIOEvent:
     data = json.loads(message.removeprefix("42"))
-    return SocketIOEvent(name=data[0], obj=json.loads(data[1]))
+    return SocketIOEvent(name=data[0], obj=data[1])
 
 
 def retrieve_project_state_from_decoded_message(event: SocketIOEvent) -> RunningState:
diff --git a/services/static-webserver/client/source/class/osparc/WatchDog.js b/services/static-webserver/client/source/class/osparc/WatchDog.js
@@ -48,11 +48,9 @@ qx.Class.define("osparc.WatchDog", {
 
     // register for socket.io event to change the default heartbeat interval
     const socket = osparc.wrapper.WebSocket.getInstance();
-    const socketIoEventName = "set_heartbeat_emit_interval";
-    socket.removeSlot(socketIoEventName);
-    socket.on(socketIoEventName, jsonString => {
-      const data = JSON.parse(jsonString);
-      const newInterval = parseInt(data.interval) * 1000;
+    socket.removeSlot("set_heartbeat_emit_interval");
+    socket.on("set_heartbeat_emit_interval", ({ interval }) => {
+      const newInterval = parseInt(interval) * 1000;
       this.setHeartbeatInterval(newInterval);
     }, this);
   },
diff --git a/services/static-webserver/client/source/class/osparc/dashboard/StudyBrowser.js b/services/static-webserver/client/source/class/osparc/dashboard/StudyBrowser.js
@@ -341,13 +341,11 @@ qx.Class.define("osparc.dashboard.StudyBrowser", {
 
     __attachEventHandlers: function() {
       const socket = osparc.wrapper.WebSocket.getInstance();
-      const slotName = "projectStateUpdated";
-      socket.on(slotName, jsonString => {
-        const data = JSON.parse(jsonString);
+      socket.on("projectStateUpdated", data => {
         if (data) {
           const studyId = data["project_uuid"];
-          const state = ("data" in data) ? data["data"] : {};
-          const errors = ("errors" in data) ? data["errors"] : [];
+          const state = ("data" in data) ? data.data : {};
+          const errors = ("errors" in data) ? data.errors : [];
           this.__studyStateReceived(studyId, state, errors);
         }
       }, this);
diff --git a/services/static-webserver/client/source/class/osparc/dashboard/TemplateBrowser.js b/services/static-webserver/client/source/class/osparc/dashboard/TemplateBrowser.js
@@ -66,13 +66,11 @@ qx.Class.define("osparc.dashboard.TemplateBrowser", {
 
     __attachEventHandlers: function() {
       const socket = osparc.wrapper.WebSocket.getInstance();
-      const slotName = "projectStateUpdated";
-      socket.on(slotName, jsonString => {
-        const data = JSON.parse(jsonString);
+      socket.on("projectStateUpdated", data => {
         if (data) {
           const templateId = data["project_uuid"];
-          const state = ("data" in data) ? data["data"] : {};
-          const errors = ("errors" in data) ? data["errors"] : [];
+          const state = ("data" in data) ? data.data : {};
+          const errors = ("errors" in data) ? data.errors : [];
           this.__templateStateReceived(templateId, state, errors);
         }
       }, this);
diff --git a/services/static-webserver/client/source/class/osparc/desktop/MainPage.js b/services/static-webserver/client/source/class/osparc/desktop/MainPage.js
@@ -91,10 +91,8 @@ qx.Class.define("osparc.desktop.MainPage", {
 
     __listenToWalletSocket: function() {
       const socket = osparc.wrapper.WebSocket.getInstance();
-      const slotName = "walletOsparcCreditsUpdated";
-      if (!socket.slotExists(slotName)) {
-        socket.on(slotName, jsonString => {
-          const data = JSON.parse(jsonString);
+      if (!socket.slotExists("walletOsparcCreditsUpdated")) {
+        socket.on("walletOsparcCreditsUpdated", data => {
           const store = osparc.store.Store.getInstance();
           const walletFound = store.getWallets().find(wallet => wallet.getWalletId() === parseInt(data["wallet_id"]));
           if (walletFound) {
diff --git a/services/static-webserver/client/source/class/osparc/desktop/MainPageDesktop.js b/services/static-webserver/client/source/class/osparc/desktop/MainPageDesktop.js
@@ -78,10 +78,8 @@ qx.Class.define("osparc.desktop.MainPageDesktop", {
   members: {
     __listenToWalletSocket: function() {
       const socket = osparc.wrapper.WebSocket.getInstance();
-      const slotName = "walletOsparcCreditsUpdated";
-      if (!socket.slotExists(slotName)) {
-        socket.on(slotName, jsonString => {
-          const data = JSON.parse(jsonString);
+      if (!socket.slotExists("walletOsparcCreditsUpdated")) {
+        socket.on("walletOsparcCreditsUpdated", data => {
           const store = osparc.store.Store.getInstance();
           const walletFound = store.getWallets().find(wallet => wallet.getWalletId() === parseInt(data["wallet_id"]));
           if (walletFound) {
diff --git a/services/static-webserver/client/source/class/osparc/desktop/StudyEditor.js b/services/static-webserver/client/source/class/osparc/desktop/StudyEditor.js
@@ -404,9 +404,8 @@ qx.Class.define("osparc.desktop.StudyEditor", {
           osparc.store.Store.getInstance().getStudyState(pipelineId);
         }, 60000);
         const socket = osparc.wrapper.WebSocket.getInstance();
-        socket.getSocket().once("projectStateUpdated", jsonStr => {
-          const study = JSON.parse(jsonStr);
-          if (study["project_uuid"] === pipelineId) {
+        socket.getSocket().once("projectStateUpdated", ({ "project_uuid": projectUuid }) => {
+          if (projectUuid === pipelineId) {
             clearTimeout(timer);
           }
         });
diff --git a/services/static-webserver/client/source/class/osparc/desktop/WorkbenchView.js b/services/static-webserver/client/source/class/osparc/desktop/WorkbenchView.js
@@ -656,22 +656,20 @@ qx.Class.define("osparc.desktop.WorkbenchView", {
       const socket = osparc.wrapper.WebSocket.getInstance();
 
       // callback for incoming logs
-      const slotName = "logger";
-      if (!socket.slotExists(slotName)) {
-        socket.on(slotName, jsonString => {
-          const data = JSON.parse(jsonString);
+      if (!socket.slotExists("logger")) {
+        socket.on("logger", data => {
           if (Object.prototype.hasOwnProperty.call(data, "project_id") && this.getStudy().getUuid() !== data["project_id"]) {
             // Filtering out logs from other studies
             return;
           }
           const nodeId = data["node_id"];
-          const messages = data["messages"];
+          const messages = data.messages;
           const logLevelMap = osparc.widget.logger.LoggerView.LOG_LEVEL_MAP;
           const logLevel = ("log_level" in data) ? logLevelMap[data["log_level"]] : "INFO";
           this.__logsToLogger(nodeId, messages, logLevel);
         }, this);
       }
-      socket.emit(slotName);
+      socket.emit("logger");
 
       // callback for incoming progress
       const slotName2 = "progress";
@@ -699,20 +697,16 @@ qx.Class.define("osparc.desktop.WorkbenchView", {
       this.listenToNodeProgress();
 
       // callback for events
-      const slotName3 = "event";
-      if (!socket.slotExists(slotName3)) {
-        socket.on(slotName3, jsonString => {
-          const data = JSON.parse(jsonString);
+      if (!socket.slotExists("event")) {
+        socket.on("event", data => {
+          const { action, "node_id": nodeId } = data
           if (Object.prototype.hasOwnProperty.call(data, "project_id") && this.getStudy().getUuid() !== data["project_id"]) {
             // Filtering out logs from other studies
             return;
           }
-          const action = data["action"];
           if (action == "RELOAD_IFRAME") {
             // TODO: maybe reload iframe in the future
             // for now a message is displayed to the user
-            const nodeId = data["node_id"];
-
             const workbench = this.getStudy().getWorkbench();
             const node = workbench.getNode(nodeId);
             const label = node.getLabel();
@@ -726,10 +720,8 @@ qx.Class.define("osparc.desktop.WorkbenchView", {
     listenToNodeUpdated: function() {
       const socket = osparc.wrapper.WebSocket.getInstance();
 
-      const slotName = "nodeUpdated";
-      if (!socket.slotExists(slotName)) {
-        socket.on(slotName, jsonString => {
-          const data = JSON.parse(jsonString);
+      if (!socket.slotExists("nodeUpdated")) {
+        socket.on("nodeUpdated", data => {
           this.getStudy().nodeUpdated(data);
         }, this);
       }
@@ -738,10 +730,8 @@ qx.Class.define("osparc.desktop.WorkbenchView", {
     listenToNodeProgress: function() {
       const socket = osparc.wrapper.WebSocket.getInstance();
 
-      const slotName = "nodeProgress";
-      if (!socket.slotExists(slotName)) {
-        socket.on(slotName, jsonString => {
-          const data = JSON.parse(jsonString);
+      if (!socket.slotExists("nodeProgress")) {
+        socket.on("nodeProgress", data => {
           this.getStudy().nodeNodeProgressSequence(data);
         }, this);
       }
diff --git a/services/static-webserver/client/source/class/osparc/desktop/credits/BuyCreditsStepper.js b/services/static-webserver/client/source/class/osparc/desktop/credits/BuyCreditsStepper.js
@@ -41,8 +41,7 @@ qx.Class.define("osparc.desktop.credits.BuyCreditsStepper", {
           osparc.data.Resources.fetch("payments", "payWithPaymentMethod", params)
             .then(data => {
               const { paymentId } = data
-              osparc.wrapper.WebSocket.getInstance().getSocket().once("paymentCompleted", wsData => {
-                const paymentData = JSON.parse(wsData);
+              osparc.wrapper.WebSocket.getInstance().getSocket().once("paymentCompleted", paymentData => {
                 if (paymentId === paymentData.paymentId) {
                   this.__paymentCompleted(paymentData)
                   this.__form.setFetching(false);
@@ -62,8 +61,7 @@ qx.Class.define("osparc.desktop.credits.BuyCreditsStepper", {
                 decorator: "no-border-2"
               });
               this.add(this.__iframe);
-              osparc.wrapper.WebSocket.getInstance().getSocket().once("paymentCompleted", wsData => {
-                const paymentData = JSON.parse(wsData);
+              osparc.wrapper.WebSocket.getInstance().getSocket().once("paymentCompleted", paymentData => {
                 if (paymentId === paymentData.paymentId) {
                   this.__paymentCompleted(paymentData);
                 }
diff --git a/services/static-webserver/client/source/class/osparc/desktop/paymentMethods/PaymentMethods.js b/services/static-webserver/client/source/class/osparc/desktop/paymentMethods/PaymentMethods.js
@@ -95,8 +95,7 @@ qx.Class.define("osparc.desktop.paymentMethods.PaymentMethods", {
         osparc.data.Resources.fetch("paymentMethods", "init", params)
           .then(data => {
             const gatewayWindow = this.__popUpPaymentGateway(data.paymentMethodId, data.paymentMethodFormUrl);
-            osparc.wrapper.WebSocket.getInstance().getSocket().once("paymentMethodAcknowledged", wsData => {
-              const {paymentMethodId} = JSON.parse(wsData);
+            osparc.wrapper.WebSocket.getInstance().getSocket().once("paymentMethodAcknowledged", ({ paymentMethodId }) => {
               if (paymentMethodId === data.paymentMethodId) {
                 gatewayWindow.close();
                 this.__fetchPaymentMethods();
diff --git a/services/static-webserver/client/source/class/osparc/snapshots/IterationsView.js b/services/static-webserver/client/source/class/osparc/snapshots/IterationsView.js
@@ -119,14 +119,12 @@ qx.Class.define("osparc.snapshots.IterationsView", {
 
     __listenToNodeUpdates: function() {
       const socket = osparc.wrapper.WebSocket.getInstance();
-      const slotName = "nodeUpdated";
-      socket.on(slotName, data => {
-        const dataUpdate = JSON.parse(data);
-        const idx = this.__iterations.findIndex(it => it["uuid"] === dataUpdate["project_id"]);
+      socket.on("nodeUpdated", data => {
+        const idx = this.__iterations.findIndex(it => it.uuid === data["project_id"]);
         if (idx === -1) {
           return;
         }
-        this.__iterationUpdated(dataUpdate);
+        this.__iterationUpdated(data);
       }, this);
     },
 
@@ -297,9 +295,7 @@ qx.Class.define("osparc.snapshots.IterationsView", {
     },
 
     unlistenToNodeUpdates: function() {
-      const socket = osparc.wrapper.WebSocket.getInstance();
-      const slotName = "nodeUpdated";
-      socket.removeSlot(slotName);
+      osparc.wrapper.WebSocket.getInstance().removeSlot("nodeUpdated");
     }
   }
 });
diff --git a/services/web/server/src/simcore_service_webserver/projects/_observer.py b/services/web/server/src/simcore_service_webserver/projects/_observer.py
@@ -25,8 +25,10 @@ async def _on_user_disconnected(
     user_id: int,
     client_session_id: str,
     app: web.Application,
-    product_name: ProductName,  # pylint: disable=unused-argument
+    product_name: ProductName,
 ) -> None:
+    assert product_name  # nosec
+
     # check if there is a project resource
     with managed_resource(user_id, client_session_id, app) as user_session:
         projects: list[str] = await user_session.find(PROJECT_ID_KEY)
diff --git a/services/web/server/src/simcore_service_webserver/socketio/_handlers.py b/services/web/server/src/simcore_service_webserver/socketio/_handlers.py
@@ -69,6 +69,7 @@ async def _handler(request: web.Request) -> tuple[UserID, ProductName]:
             socketio_session["user_id"] = user_id
             socketio_session["client_session_id"] = client_session_id
             socketio_session["request"] = request
+            socketio_session["product_name"] = product.name
 
         # REDIS wrapper
         with managed_resource(user_id, client_session_id, app) as resource_registry:
@@ -165,6 +166,7 @@ async def disconnect(socket_id: SocketID, app: web.Application) -> None:
     async with sio.session(socket_id) as socketio_session:
         if user_id := socketio_session.get("user_id"):
             client_session_id = socketio_session["client_session_id"]
+            product_name = socketio_session["product_name"]
 
             with log_context(
                 _logger,
@@ -182,7 +184,7 @@ async def disconnect(socket_id: SocketID, app: web.Application) -> None:
                     user_id,
                     client_session_id,
                     app,
-                    "s4l",  # NOTE: will be changed after https://github.com/ITISFoundation/osparc-simcore/issues/4776
+                    product_name,
                 )
 
         else:
diff --git a/services/web/server/src/simcore_service_webserver/socketio/messages.py b/services/web/server/src/simcore_service_webserver/socketio/messages.py
@@ -9,7 +9,7 @@
 from models_library.api_schemas_webserver.socketio import SocketIORoomStr
 from models_library.socketio import SocketMessageDict
 from models_library.users import GroupID, UserID
-from servicelib.json_serialization import json_dumps
+from models_library.utils.fastapi_encoders import jsonable_encoder
 from socketio import AsyncServer
 
 from ._utils import get_socket_server
@@ -44,7 +44,7 @@ async def _safe_emit(
     # NOTE 2: `emit` method is not designed to be used concurrently
     try:
         event = message["event_type"]
-        data = json_dumps(message["data"])
+        data = jsonable_encoder(message["data"])
         await sio.emit(
             event=event,
             data=data,
diff --git a/services/web/server/tests/integration/02/notifications/test_rabbitmq_consumers.py b/services/web/server/tests/integration/02/notifications/test_rabbitmq_consumers.py
@@ -4,7 +4,6 @@
 # pylint: disable=too-many-arguments
 
 import asyncio
-import json
 from collections.abc import Awaitable, Callable
 from random import choice
 from typing import Any
@@ -125,7 +124,7 @@ async def _assert_handler_called_with_json(
             )
             handler.assert_called_once()
             call_args, _call_kwargs = handler.call_args
-            assert json.loads(call_args[0]) == expected_call
+            assert call_args[0] == expected_call
             print(f"calls received! {attempt.retry_state.retry_object.statistics}")
 
 
diff --git a/services/web/server/tests/unit/with_dbs/02/test_projects_states_handlers.py b/services/web/server/tests/unit/with_dbs/02/test_projects_states_handlers.py
@@ -5,7 +5,6 @@
 # pylint: disable=unused-variable
 
 import asyncio
-import json
 import time
 from collections.abc import Awaitable, Callable, Iterator
 from copy import deepcopy
@@ -37,6 +36,7 @@
     ServiceResourcesDict,
     ServiceResourcesDictHelpers,
 )
+from models_library.utils.fastapi_encoders import jsonable_encoder
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from pytest_simcore.helpers.utils_assert import assert_status
 from pytest_simcore.helpers.utils_login import UserInfoDict, log_client_in
@@ -217,7 +217,7 @@ async def _assert_project_state_updated(
 
         calls = [
             call(
-                json.dumps(
+                jsonable_encoder(
                     {
                         "project_uuid": shared_project["uuid"],
                         "data": p_state.dict(by_alias=True, exclude_unset=True),
diff --git a/services/web/server/tests/unit/with_dbs/03/garbage_collector/test_resource_manager.py b/services/web/server/tests/unit/with_dbs/03/garbage_collector/test_resource_manager.py
@@ -5,7 +5,6 @@
 
 
 import asyncio
-import json
 import logging
 from asyncio import Future
 from collections.abc import AsyncIterator, Awaitable, Callable
@@ -22,6 +21,7 @@
 from aiohttp import web
 from aiohttp.test_utils import TestClient
 from aioresponses import aioresponses
+from models_library.utils.fastapi_encoders import jsonable_encoder
 from pytest_mock import MockerFixture
 from pytest_simcore.helpers.utils_assert import assert_status
 from pytest_simcore.helpers.utils_envs import setenvs_from_dict
@@ -586,7 +586,7 @@ async def test_interactive_services_remain_after_websocket_reconnection_from_2_t
     ):
         with attempt:
             socket_project_state_update_mock_callable.assert_called_with(
-                json.dumps(
+                jsonable_encoder(
                     {
                         "project_uuid": empty_user_project["uuid"],
                         "data": {
