diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/manager.py b/trustgraph-flow/trustgraph/gateway/dispatch/manager.py index 48240532..44305d2f 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/manager.py +++ b/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): diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/socket.py b/trustgraph-flow/trustgraph/gateway/endpoint/socket.py index c7b194be..1bfec637 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/socket.py +++ b/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) )