Multiple streams working

trustgraph-flow/trustgraph/gateway/dispatch/manager.py

@@ -1,5 +1,6 @@
 
 import asyncio
+import uuid
 
 from . embeddings import EmbeddingsRequestor
 from . agent import AgentRequestor
@@ -27,9 +28,9 @@ request_response_dispatchers = {
 }
 
 receive_dispatchers = {
-    "triples-store": TriplesStream,
-    "graph-embeddings-store": GraphEmbeddingsStream,
-    "document-embeddings-store": DocumentEmbeddingsStream,
+    "triples": TriplesStream,
+    "graph-embeddings": GraphEmbeddingsStream,
+    "document-embeddings": DocumentEmbeddingsStream,
 }
 
 class TestDispatcher:
@@ -130,9 +131,6 @@ class DispatcherManager:
         flow = params.get("flow")
         kind = params.get("kind")
 
-        # FIXME: What?!?!
-        kind += "-store"
-
         if flow not in self.flows:
             raise RuntimeError("Invalid flow")
 
@@ -141,28 +139,25 @@ class DispatcherManager:
 
         key = (flow, kind)
 
-        if key in self.dispatchers:
-            return self.dispatchers[key]
-
         intf_defs = self.flows[flow]["interfaces"]
 
         if kind not in intf_defs:
             raise RuntimeError("This kind not supported by flow")
 
-        qconfig = intf_defs[kind]
+        # FIXME: The -store bit, does it make sense?
+        qconfig = intf_defs[kind + "-store"]
 
+        id = str(uuid.uuid4())
         dispatcher = receive_dispatchers[kind](
             pulsar_client = self.pulsar_client,
             ws = ws,
             running = running,
             # FIXME!
             queue = qconfig,
-            consumer = f"api-gateway-{flow}-{kind}-request",
-            subscriber = f"api-gateway-{flow}-{kind}-request",
+            consumer = f"api-gateway-{id}",
+            subscriber = f"api-gateway-{id}",
         )
 
-        self.dispatchers[key] = dispatcher
-
         return dispatcher
 
     async def process(self, data, responder, params):

trustgraph-flow/trustgraph/gateway/endpoint/socket.py

@@ -26,13 +26,10 @@ class SocketEndpoint:
 
     async def listener(self, ws, dispatcher, running):
 
-        print("In listener...!!!!!!!!!!!!!!!!!!!!")
         async for msg in ws:
 
             # On error, finish
-            print ("mt:", msg.type)
             if msg.type == WSMsgType.TEXT:
-                print("YER!!!!!")
                 await dispatcher.receive(msg)
                 continue
             elif msg.type == WSMsgType.BINARY:
@@ -65,17 +62,14 @@ class SocketEndpoint:
 
                 running = Running()
 
-                print("Create...")
                 dispatcher = await self.dispatcher(
                     ws, running, request.match_info
                 )
 
-                print("Create worker...")
                 worker_task = tg.create_task(
                     self.worker(ws, dispatcher, running)
                 )
 
-                print("Create listener")
                 lsnr_task = tg.create_task(
                     self.listener(ws, dispatcher, running)
                 )