review @GitHK

Author: matusdrobuliak66

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

@@ -38,6 +38,8 @@
 _logger = logging.getLogger(__name__)
 
 _APP_RABBITMQ_CONSUMERS_KEY: Final[str] = f"{__name__}.rabbit_consumers"
+APP_WALLET_SUBSCRIPTIONS_KEY = "wallet_subscriptions"
+APP_WALLET_SUBSCRIPTION_LOCK_KEY = "wallet_subscription_lock"
 
 
 async def _convert_to_node_update_event(
@@ -195,8 +197,10 @@ async def on_cleanup_ctx_rabbitmq_consumers(
         app, _EXCHANGE_TO_PARSER_CONFIG
     )
 
-    app["wallet_subscriptions"] = defaultdict(int)  # wallet_id -> subscriber count
-    app["wallet_subscription_lock"] = asyncio.Lock()  # For thread-safe operations
+    app[APP_WALLET_SUBSCRIPTIONS_KEY] = defaultdict(
+        int
+    )  # wallet_id -> subscriber count
+    app[APP_WALLET_SUBSCRIPTION_LOCK_KEY] = asyncio.Lock()  # For thread-safe operations
 
     yield
 

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

@@ -7,15 +7,19 @@
 from servicelib.rabbitmq import RabbitMQClient
 
 from ..rabbitmq import get_rabbitmq_client
+from ._rabbitmq_exclusive_queue_consumers import (
+    APP_WALLET_SUBSCRIPTION_LOCK_KEY,
+    APP_WALLET_SUBSCRIPTIONS_KEY,
+)
 
 _logger = logging.getLogger(__name__)
 
 
 async def subscribe(app: web.Application, wallet_id: WalletID) -> None:
 
-    async with app["wallet_subscription_lock"]:
-        counter = app["wallet_subscriptions"][wallet_id]
-        app["wallet_subscriptions"][wallet_id] += 1
+    async with app[APP_WALLET_SUBSCRIPTION_LOCK_KEY]:
+        counter = app[APP_WALLET_SUBSCRIPTIONS_KEY][wallet_id]
+        app[APP_WALLET_SUBSCRIPTIONS_KEY][wallet_id] += 1
 
         if counter == 0:  # First subscriber
             rabbit_client: RabbitMQClient = get_rabbitmq_client(app)
@@ -26,10 +30,10 @@ async def subscribe(app: web.Application, wallet_id: WalletID) -> None:
 
 async def unsubscribe(app: web.Application, wallet_id: WalletID) -> None:
 
-    async with app["wallet_subscription_lock"]:
-        counter = app["wallet_subscriptions"].get(wallet_id, 0)
+    async with app[APP_WALLET_SUBSCRIPTION_LOCK_KEY]:
+        counter = app[APP_WALLET_SUBSCRIPTIONS_KEY].get(wallet_id, 0)
         if counter > 0:
-            app["wallet_subscriptions"][wallet_id] -= 1
+            app[APP_WALLET_SUBSCRIPTIONS_KEY][wallet_id] -= 1
 
             if counter == 1:  # Last subscriber
                 rabbit_client: RabbitMQClient = get_rabbitmq_client(app)