worker: add basic _dispatch_work logic

blueprints/streaming.py

@@ -13,6 +13,7 @@ log = logging.getLogger(__name__)
 class OP:
     SUBSCRIBED = 1
 
+
 class ErrorCodes:
     TOO_MUCH = 4420
 

elstat/manager.py

@@ -1,6 +1,6 @@
 import logging
 
-from typing import List
+from typing import List, Dict, Any
 
 from .consts import ADAPTERS
 from .worker import ServiceWorker
@@ -40,9 +40,8 @@ class ServiceManager:
         );
         """)
 
-    def _check(self, columns: tuple, field: str, chan_name: str = None):
-        if chan_name is None:
-            chan_name = field
+    def _check(self, columns: tuple, field: str, worker_name: str):
+        chan_name = f'{field}:{worker_name}'
 
         if field in columns and chan_name not in self.subscribers:
             self.subscribers[chan_name] = []
@@ -51,8 +50,8 @@ class ServiceManager:
     def _create_channels(self, worker):
         columns = worker.adapter.spec['db']
 
-        self._check(columns, 'status')
-        self._check(columns, 'latency', f'latency:{worker.name}')
+        self._check(columns, 'status', worker.name)
+        self._check(columns, 'latency', worker.name)
 
     def _start(self):
         for name, service in self.cfg.SERVICES.items():
@@ -95,3 +94,9 @@ class ServiceManager:
 
         log.info(f'unsubscribed {websocket.client_id} from {unsub}')
         return unsub
+
+    def publish(self, channel: str, data: Any):
+        pass
+
+    def publish_many(self, data: Dict[str, Any]):
+        pass

elstat/worker.py

@@ -36,6 +36,18 @@ class ServiceWorker:
         conn.execute(query, (timestamp,) + result)
         conn.commit()
 
+        await self._dispatch_work(columns, result)
+
+    async def _dispatch_work(self, columns, result: tuple):
+        prechan = columns[1:]
+        chans = [f'{chan}:{self.name}' for chan in prechan]
+
+        if len(chans) > 1:
+            self.manager.publish_many({
+                chan: result[idx] for idx, chan in enumerate(chans)})
+        else:
+            self.manager.publish(chans[0], result)
+
     async def _work_loop(self):
         try:
             while True: