Merge pull request #381 from trustgraph-ai/fix/import-queues-not-working

Fix missing queue initialisation

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

@@ -22,6 +22,9 @@ class DocumentEmbeddingsImport:
             pulsar_client, topic = queue, schema = DocumentEmbeddings
         )
 
+    async def start(self):
+        await self.publisher.start()
+
     async def destroy(self):
         self.running.stop()
 

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

@@ -22,6 +22,9 @@ class GraphEmbeddingsImport:
             pulsar_client, topic = queue, schema = GraphEmbeddings
         )
 
+    async def start(self):
+        await self.publisher.start()
+
     async def destroy(self):
         self.running.stop()
 

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

@@ -160,6 +160,8 @@ class DispatcherManager:
             queue = qconfig,
         )
 
+        await dispatcher.start()
+
         return dispatcher
 
     async def process_flow_export(self, ws, running, params):

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

@@ -22,6 +22,9 @@ class TriplesImport:
             pulsar_client, topic = queue, schema = Triples
         )
 
+    async def start(self):
+        await self.publisher.start()
+
     async def destroy(self):
         self.running.stop()