From e9ae0b27f387b65532bdf15a756195a03ce02037 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Fri, 2 May 2025 14:51:34 +0100 Subject: [PATCH] Dispatch manager --- .../trustgraph/gateway/dispatch/manager.py | 76 +++++++++++++ .../trustgraph/gateway/endpoint/flows.py | 104 +----------------- 2 files changed, 81 insertions(+), 99 deletions(-) create mode 100644 trustgraph-flow/trustgraph/gateway/dispatch/manager.py diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/manager.py b/trustgraph-flow/trustgraph/gateway/dispatch/manager.py new file mode 100644 index 00000000..01352524 --- /dev/null +++ b/trustgraph-flow/trustgraph/gateway/dispatch/manager.py @@ -0,0 +1,76 @@ + +import asyncio + +class TestDispatcher: + def __init__(self, pulsar_client, timeout=120): + self.pulsar_client = pulsar_client + timeout = timeout + async def process(self, data, responder): + result = { "result": "Hello world!" } + + if responder: + await responder(result, True) + + return result + +class TestDispatcher2: + def __init__(self, pulsar_client, timeout=120): + self.pulsar_client = pulsar_client + timeout = timeout + async def process(self, data, responder, params): + + thing = params['thing'] + + result = { "result": "Hello world!!", "thing": thing } + + if responder: + await responder(result, True) + + return result + +class TestDispatcher3: + def __init__(self, pulsar_client, timeout=120): + self.pulsar_client = pulsar_client + self.timeout = timeout + + async def dispatch(self, ws, running, request): + + class Runner: + def __init__(self, ws, running): + self.ws = ws + self.running = running + + async def destroy(self): + + if self.ws: + await self.ws.close() + self.ws = None + + self.running.stop() + + async def run(self): + + i = 0 + + while self.running.get(): + await self.ws.send_json({"i": i}) + i += 1 + await asyncio.sleep(1) + + await self.ws.close() + self.ws = None + + async def receive(self, msg): + print("Receive:", msg.data) + + return Runner(ws, running) + +class DispatcherManager: + def __init__(self, pulsar_client): + self.pulsar_client = pulsar_client + def dispatch_test_service(self): + return TestDispatcher(pulsar_client = self.pulsar_client) + def dispatch_flow_service(self): + return TestDispatcher2(pulsar_client = self.pulsar_client) + def dispatch_socket_service(self): + return TestDispatcher3(pulsar_client = self.pulsar_client).dispatch diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/flows.py b/trustgraph-flow/trustgraph/gateway/endpoint/flows.py index 8deef287..a7a06000 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/flows.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/flows.py @@ -9,18 +9,7 @@ from . constant_endpoint import ConstantEndpoint from . variable_endpoint import VariableEndpoint from . socket import SocketEndpoint -#from . flow_endpoint import FlowEndpoint - -# from .. dispatch.agent import AgentRequestor -# from .. dispatch.text_completion import TextCompletionRequestor -# from .. dispatch.prompt import PromptRequestor -# from .. dispatch.graph_rag import GraphRagRequestor -# from .. dispatch.document_rag import DocumentRagRequestor -# from .. dispatch.triples_query import TriplesQueryRequestor -# from .. dispatch.embeddings import EmbeddingsRequestor -# from .. dispatch.graph_embeddings_query import GraphEmbeddingsQueryRequestor -# from .. dispatch.prompt import PromptRequestor -# from .. dispatch.triples_stream import TriplesStream +from .. dispatch.manager import DispatcherManager class FlowEndpointManager: @@ -33,111 +22,28 @@ class FlowEndpointManager: self.services = { } - class Dispatcher: - def __init__(self, mode, name, timeout=timeout): - self.mode = mode - self.name = name - self.pulsar_client = pulsar_client - timeout = timeout - async def process(self, data, responder): - result = { "result": "Hello world" } - - if responder: - await responder(result, True) - - return result - - class Dispatcher2: - def __init__(self, mode, name, timeout=timeout): - self.mode = mode - self.name = name - self.pulsar_client = pulsar_client - timeout = timeout - async def process(self, data, responder, params): - - thing = params['thing'] - - result = { "result": "Hello world", "thing": thing } - - if responder: - await responder(result, True) - - return result - - class Dispatcher3: - def __init__(self, mode, name, timeout=timeout): - self.mode = mode - self.name = name - self.pulsar_client = pulsar_client - timeout = timeout - - async def dispatch(self, ws, running, request): - - class Runner: - def __init__(self, ws, running): - self.ws = ws - self.running = running - - async def destroy(self): - - if self.ws: - await self.ws.close() - self.ws = None - - self.running.stop() - - async def run(self): - - i = 0 - - while self.running.get(): - await self.ws.send_json({"i": i}) - i += 1 - await asyncio.sleep(1) - - await self.ws.close() - self.ws = None - - async def receive(self, msg): - print("Receive:", msg.data) - - return Runner(ws, running) + dm = DispatcherManager(pulsar_client) self.endpoints = [ ConstantEndpoint( endpoint_path = "/api/v1/test", auth = auth, - dispatcher = Dispatcher(None, "test") + dispatcher = dm.dispatch_test_service(), ), VariableEndpoint( endpoint_path = "/api/v1/test/{thing}", auth = auth, - dispatcher = Dispatcher2(None, "test") + dispatcher = dm.dispatch_flow_service(), ), SocketEndpoint( endpoint_path = "/api/v1/test2", auth = auth, - dispatcher = Dispatcher3(None, "test2").dispatch, + dispatcher = dm.dispatch_socket_service() ), ] self.config_receiver.add_handler(self) - async def ASDcreate_stream_dispatch(self, ws, running, request): - - flow_id = request.match_info['flow'] - kind = request.match_info['kind'] - k = (flow_id, kind) - - print("Service", k) - - print(self.services) - - if k not in self.services: - raise web.HTTPBadRequest() - - raise RuntimeError("Not impl") - def add_routes(self, app): for ep in self.endpoints: ep.add_routes(app)