From 0bd1a7265eb933f76e12f4662de834861cdd8733 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Tue, 3 Dec 2024 16:15:31 +0000 Subject: [PATCH] Back out previous change --- .../trustgraph/api/gateway/endpoint.py | 11 ++--- .../api/gateway/graph_embeddings_load.py | 4 +- .../api/gateway/graph_embeddings_stream.py | 4 +- .../trustgraph/api/gateway/publisher.py | 26 ++++++------ .../trustgraph/api/gateway/service.py | 33 +++------------ .../trustgraph/api/gateway/socket.py | 6 --- .../trustgraph/api/gateway/subscriber.py | 42 ++++++++++--------- .../trustgraph/api/gateway/triples_load.py | 4 +- .../trustgraph/api/gateway/triples_stream.py | 4 +- 9 files changed, 52 insertions(+), 82 deletions(-) diff --git a/trustgraph-flow/trustgraph/api/gateway/endpoint.py b/trustgraph-flow/trustgraph/api/gateway/endpoint.py index c7cd6b04..dc380f4b 100644 --- a/trustgraph-flow/trustgraph/api/gateway/endpoint.py +++ b/trustgraph-flow/trustgraph/api/gateway/endpoint.py @@ -41,15 +41,10 @@ class ServiceEndpoint: self.operation = "service" - async def start(self, client): + async def start(self): - self.pub_task = asyncio.create_task(self.pub.run(client)) - self.sub_task = asyncio.create_task(self.sub.run(client)) - - async def join(self): - - await self.pub_task - await self.sub_task + self.pub_task = asyncio.create_task(self.pub.run()) + self.sub_task = asyncio.create_task(self.sub.run()) def add_routes(self, app): diff --git a/trustgraph-flow/trustgraph/api/gateway/graph_embeddings_load.py b/trustgraph-flow/trustgraph/api/gateway/graph_embeddings_load.py index 764e7210..15efdf5b 100644 --- a/trustgraph-flow/trustgraph/api/gateway/graph_embeddings_load.py +++ b/trustgraph-flow/trustgraph/api/gateway/graph_embeddings_load.py @@ -29,10 +29,10 @@ class GraphEmbeddingsLoadEndpoint(SocketEndpoint): schema=JsonSchema(GraphEmbeddings) ) - async def start(self, client): + async def start(self): self.task = asyncio.create_task( - self.publisher.run(client) + self.publisher.run() ) async def listener(self, ws, running): diff --git a/trustgraph-flow/trustgraph/api/gateway/graph_embeddings_stream.py b/trustgraph-flow/trustgraph/api/gateway/graph_embeddings_stream.py index 12647547..7f3e5e18 100644 --- a/trustgraph-flow/trustgraph/api/gateway/graph_embeddings_stream.py +++ b/trustgraph-flow/trustgraph/api/gateway/graph_embeddings_stream.py @@ -28,10 +28,10 @@ class GraphEmbeddingsStreamEndpoint(SocketEndpoint): schema=JsonSchema(GraphEmbeddings) ) - async def start(self, client): + async def start(self): self.task = asyncio.create_task( - self.subscriber.run(client) + self.subscriber.run() ) async def async_thread(self, ws, running): diff --git a/trustgraph-flow/trustgraph/api/gateway/publisher.py b/trustgraph-flow/trustgraph/api/gateway/publisher.py index 2bbf05d9..1bff44dd 100644 --- a/trustgraph-flow/trustgraph/api/gateway/publisher.py +++ b/trustgraph-flow/trustgraph/api/gateway/publisher.py @@ -1,5 +1,6 @@ import asyncio +import aiopulsar class Publisher: @@ -11,23 +12,24 @@ class Publisher: self.q = asyncio.Queue(maxsize=max_size) self.chunking_enabled = chunking_enabled - async def run(self, client): + async def run(self): while True: try: - async with client.create_producer( - topic=self.topic, - schema=self.schema, - chunking_enabled=self.chunking_enabled, - ) as producer: - while True: - id, item = await self.q.get() + async with aiopulsar.connect(self.pulsar_host) as client: + async with client.create_producer( + topic=self.topic, + schema=self.schema, + chunking_enabled=self.chunking_enabled, + ) as producer: + while True: + id, item = await self.q.get() - if id: - await producer.send(item, { "id": id }) - else: - await producer.send(item) + if id: + await producer.send(item, { "id": id }) + else: + await producer.send(item) except Exception as e: print("Exception:", e, flush=True) diff --git a/trustgraph-flow/trustgraph/api/gateway/service.py b/trustgraph-flow/trustgraph/api/gateway/service.py index 38a86a51..a25dd9dc 100755 --- a/trustgraph-flow/trustgraph/api/gateway/service.py +++ b/trustgraph-flow/trustgraph/api/gateway/service.py @@ -17,7 +17,6 @@ from aiohttp import web import logging import os import base64 -import aiopulsar import pulsar from pulsar.schema import JsonSchema @@ -238,35 +237,13 @@ class Api: { "error": str(e) } ) - async def run_endpoints(self): - - async with aiopulsar.connect(self.pulsar_host) as client: - - for ep in self.endpoints: - await ep.start(client) - - self.doc_ingest_pub_task = asyncio.create_task( - self.document_out.run(client) - ) - - self.text_ingest_pub_task = asyncio.create_task( - self.text_out.run(client) - ) - - print("Endpoints are running...") - - # They never exit - for ep in self.endpoints: - await ep.join() - - await self.doc_ingest_pub_task - await self.text_ingest_pub_task - - print("Endpoints are stopped.") - async def app_factory(self): - self.endpoint_task = asyncio.create_task(self.run_endpoints()) + for ep in self.endpoints: + await ep.start() + + self.doc_ingest_pub_task = asyncio.create_task(self.document_out.run()) + self.text_ingest_pub_task = asyncio.create_task(self.text_out.run()) return self.app diff --git a/trustgraph-flow/trustgraph/api/gateway/socket.py b/trustgraph-flow/trustgraph/api/gateway/socket.py index a4cb0feb..869792b7 100644 --- a/trustgraph-flow/trustgraph/api/gateway/socket.py +++ b/trustgraph-flow/trustgraph/api/gateway/socket.py @@ -76,12 +76,6 @@ class SocketEndpoint: async def start(self): pass - async def join(self): - - # Nothing to wait for - while True: - await asyncio.sleep(100) - def add_routes(self, app): app.add_routes([ diff --git a/trustgraph-flow/trustgraph/api/gateway/subscriber.py b/trustgraph-flow/trustgraph/api/gateway/subscriber.py index ba53bab6..3d8840f6 100644 --- a/trustgraph-flow/trustgraph/api/gateway/subscriber.py +++ b/trustgraph-flow/trustgraph/api/gateway/subscriber.py @@ -1,5 +1,6 @@ import asyncio +import aiopulsar class Subscriber: @@ -13,32 +14,33 @@ class Subscriber: self.q = {} self.full = {} - async def run(self, client): + async def run(self): while True: try: - async with client.subscribe( - topic=self.topic, - subscription_name=self.subscription, - consumer_name=self.consumer_name, - schema=self.schema, - ) as consumer: - while True: - msg = await consumer.receive() + async with aiopulsar.connect(self.pulsar_host) as client: + async with client.subscribe( + topic=self.topic, + subscription_name=self.subscription, + consumer_name=self.consumer_name, + schema=self.schema, + ) as consumer: + while True: + msg = await consumer.receive() - # Acknowledge successful reception of the message - await consumer.acknowledge(msg) + # Acknowledge successful reception of the message + await consumer.acknowledge(msg) - try: - id = msg.properties()["id"] - except: - id = None + try: + id = msg.properties()["id"] + except: + id = None - value = msg.value() - if id in self.q: - await self.q[id].put(value) + value = msg.value() + if id in self.q: + await self.q[id].put(value) - for q in self.full.values(): - await q.put(value) + for q in self.full.values(): + await q.put(value) except Exception as e: print("Exception:", e, flush=True) diff --git a/trustgraph-flow/trustgraph/api/gateway/triples_load.py b/trustgraph-flow/trustgraph/api/gateway/triples_load.py index 0460d1e4..7f4561b1 100644 --- a/trustgraph-flow/trustgraph/api/gateway/triples_load.py +++ b/trustgraph-flow/trustgraph/api/gateway/triples_load.py @@ -27,10 +27,10 @@ class TriplesLoadEndpoint(SocketEndpoint): schema=JsonSchema(Triples) ) - async def start(self, client): + async def start(self): self.task = asyncio.create_task( - self.publisher.run(client) + self.publisher.run() ) async def listener(self, ws, running): diff --git a/trustgraph-flow/trustgraph/api/gateway/triples_stream.py b/trustgraph-flow/trustgraph/api/gateway/triples_stream.py index 571d5e61..6ecd2bdb 100644 --- a/trustgraph-flow/trustgraph/api/gateway/triples_stream.py +++ b/trustgraph-flow/trustgraph/api/gateway/triples_stream.py @@ -26,10 +26,10 @@ class TriplesStreamEndpoint(SocketEndpoint): schema=JsonSchema(Triples) ) - async def start(self, client): + async def start(self): self.task = asyncio.create_task( - self.subscriber.run(client) + self.subscriber.run() ) async def async_thread(self, ws, running):