Back out previous change

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.pub_task = asyncio.create_task(self.pub.run())
-        self.sub_task = asyncio.create_task(self.sub.run(client))
+        self.sub_task = asyncio.create_task(self.sub.run())
-
-    async def join(self):
-        
-        await self.pub_task
-        await self.sub_task
 
     def add_routes(self, app):
 

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):

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):

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(
+                async with aiopulsar.connect(self.pulsar_host) as client:
-                        topic=self.topic,
+                    async with client.create_producer(
-                        schema=self.schema,
+                            topic=self.topic,
-                        chunking_enabled=self.chunking_enabled,
+                            schema=self.schema,
-                ) as producer:
+                            chunking_enabled=self.chunking_enabled,
-                    while True:
+                    ) as producer:
-                        id, item = await self.q.get()
+                        while True:
+                            id, item = await self.q.get()
 
-                        if id:
+                            if id:
-                            await producer.send(item, { "id": id })
+                                await producer.send(item, { "id": id })
-                        else:
+                            else:
-                            await producer.send(item)
+                                await producer.send(item)
 
             except Exception as e:
                 print("Exception:", e, flush=True)

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
 

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([

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(
+                async with aiopulsar.connect(self.pulsar_host) as client:
-                    topic=self.topic,
+                    async with client.subscribe(
-                    subscription_name=self.subscription,
+                        topic=self.topic,
-                    consumer_name=self.consumer_name,
+                        subscription_name=self.subscription,
-                    schema=self.schema,
+                        consumer_name=self.consumer_name,
-                ) as consumer:
+                        schema=self.schema,
-                    while True:
+                    ) as consumer:
-                        msg = await consumer.receive()
+                        while True:
+                            msg = await consumer.receive()
 
-                        # Acknowledge successful reception of the message
+                            # Acknowledge successful reception of the message
-                        await consumer.acknowledge(msg)
+                            await consumer.acknowledge(msg)
 
-                        try:
+                            try:
-                            id = msg.properties()["id"]
+                                id = msg.properties()["id"]
-                        except:
+                            except:
-                            id = None
+                                id = None
 
-                        value = msg.value()
+                            value = msg.value()
-                        if id in self.q:
+                            if id in self.q:
-                            await self.q[id].put(value)
+                                await self.q[id].put(value)
 
-                        for q in self.full.values():
+                            for q in self.full.values():
-                            await q.put(value)
+                                await q.put(value)
 
             except Exception as e:
                 print("Exception:", e, flush=True)

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):

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):