apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-21 19:21:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Fix async/sync load issues with knowledge streaming APIs

Browse source
This commit is contained in:
Cyber MacGeddon 2025-03-15 12:14:51 +00:00
parent 88768a791b
commit 7cf9299698
5 changed files with 6 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

2
trustgraph-flow/trustgraph/gateway/document_embeddings_load.py
View file

@ -59,6 +59,6 @@ class DocumentEmbeddingsLoadEndpoint(SocketEndpoint):
], ],
) )
await self.publisher.send(None, elt) self.publisher.send(None, elt)
running.stop() running.stop()

4
trustgraph-flow/trustgraph/gateway/graph_embeddings_load.py
View file

@ -36,6 +36,7 @@ class GraphEmbeddingsLoadEndpoint(SocketEndpoint):
async def listener(self, ws, running): async def listener(self, ws, running):
async for msg in ws: async for msg in ws:
# On error, finish # On error, finish
if msg.type == WSMsgType.ERROR: if msg.type == WSMsgType.ERROR:
break break
@ -59,7 +60,6 @@ class GraphEmbeddingsLoadEndpoint(SocketEndpoint):
] ]
) )
await self.publisher.send(None, elt) self.publisher.send(None, elt)
running.stop() running.stop()

2
trustgraph-flow/trustgraph/gateway/requestor.py
View file

@ -53,11 +53,9 @@ class ServiceRequestor:
q = self.sub.subscribe(id) q = self.sub.subscribe(id)
print("BOUT TO SEDN")
await asyncio.to_thread( await asyncio.to_thread(
self.pub.send, id, self.to_request(request) self.pub.send, id, self.to_request(request)
) )
print("SENT")
while True: while True:

2
trustgraph-flow/trustgraph/gateway/socket.py
View file

@ -53,7 +53,7 @@ class SocketEndpoint:
try: try:
await self.listener(ws, running) await self.listener(ws, running)
except Exception as e: except Exception as e:
print(e, flush=True) print("Socket exception:", e, flush=True)
running.stop() running.stop()

2
trustgraph-flow/trustgraph/gateway/triples_load.py
View file

@ -51,7 +51,7 @@ class TriplesLoadEndpoint(SocketEndpoint):
triples=to_subgraph(data["triples"]), triples=to_subgraph(data["triples"]),
) )
await self.publisher.send(None, elt) self.publisher.send(None, elt)
running.stop() running.stop()