properly update

Author: sanderegg

services/web/server/src/simcore_service_webserver/notifications/_rabbitmq_exclusive_queue_consumers.py

@@ -22,12 +22,10 @@
 from servicelib.utils import logged_gather
 
 from ..projects import _projects_service
-from ..projects.exceptions import ProjectNotFoundError
 from ..rabbitmq import get_rabbitmq_client
 from ..socketio.messages import (
     SOCKET_IO_EVENT,
     SOCKET_IO_LOG_EVENT,
-    SOCKET_IO_NODE_UPDATED_EVENT,
     SOCKET_IO_WALLET_OSPARC_CREDITS_UPDATED_EVENT,
     send_message_to_standard_group,
     send_message_to_user,
@@ -43,30 +41,16 @@
 APP_WALLET_SUBSCRIPTION_LOCK_KEY: Final[str] = "wallet_subscription_lock"
 
 
-async def _convert_to_node_update_event(
+async def _notify_comp_node_progress(
     app: web.Application, message: ProgressRabbitMessageNode
-) -> SocketMessageDict | None:
-    try:
+) -> None:
+    with log_catch(_logger, reraise=False):
         project = await _projects_service.get_project_for_user(
-            app, f"{message.project_id}", message.user_id
+            app, f"{message.project_id}", message.user_id, include_state=True
+        )
+        await _projects_service.notify_project_node_update(
+            app, project, message.node_id, None
         )
-        if f"{message.node_id}" in project["workbench"]:
-            # update the project node progress with the latest value
-            project["workbench"][f"{message.node_id}"].update(
-                {"progress": round(message.report.percent_value * 100.0)}
-            )
-            return SocketMessageDict(
-                event_type=SOCKET_IO_NODE_UPDATED_EVENT,
-                data={
-                    "project_id": message.project_id,
-                    "node_id": message.node_id,
-                    "data": project["workbench"][f"{message.node_id}"],
-                },
-            )
-        _logger.warning("node not found: '%s'", message.model_dump())
-    except ProjectNotFoundError:
-        _logger.warning("project not found: '%s'", message.model_dump())
-    return None
 
 
 async def _progress_message_parser(app: web.Application, data: bytes) -> bool:
@@ -80,10 +64,9 @@ async def _progress_message_parser(app: web.Application, data: bytes) -> bool:
         message = WebSocketProjectProgress.from_rabbit_message(
             rabbit_message
         ).to_socket_dict()
-
     elif rabbit_message.progress_type is ProgressType.COMPUTATION_RUNNING:
-        message = await _convert_to_node_update_event(app, rabbit_message)
-
+        await _notify_comp_node_progress(app, rabbit_message)
+        return True
     else:
         message = WebSocketNodeProgress.from_rabbit_message(
             rabbit_message