mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-22 19:51:02 +02:00
a9197d11ee
- Keeps processing in different flows separate so that data can go to different stores / collections etc. - Potentially supports different processing flows - Tidies the processing API with common base-classes for e.g. LLMs, and automatic configuration of 'clients' to use the right queue names in a flow
69 lines
1.7 KiB
Python
69 lines
1.7 KiB
Python
|
|
from pulsar.schema import JsonSchema
|
|
import asyncio
|
|
|
|
class Producer:
|
|
|
|
def __init__(self, client, topic, schema, metrics=None):
|
|
self.client = client
|
|
self.topic = topic
|
|
self.schema = schema
|
|
|
|
self.metrics = metrics
|
|
|
|
self.running = True
|
|
self.producer = None
|
|
|
|
def __del__(self):
|
|
|
|
self.running = False
|
|
|
|
if hasattr(self, "producer"):
|
|
if self.producer:
|
|
self.producer.close()
|
|
|
|
async def start(self):
|
|
self.running = True
|
|
|
|
async def stop(self):
|
|
self.running = False
|
|
|
|
async def send(self, msg, properties={}):
|
|
|
|
if not self.running: return
|
|
|
|
while self.running and self.producer is None:
|
|
|
|
try:
|
|
print("Connect publisher to", self.topic, "...", flush=True)
|
|
self.producer = self.client.create_producer(
|
|
topic = self.topic,
|
|
schema = JsonSchema(self.schema)
|
|
)
|
|
print("Connected to", self.topic, flush=True)
|
|
except Exception as e:
|
|
print("Exception:", e, flush=True)
|
|
await asyncio.sleep(2)
|
|
|
|
if not self.running: break
|
|
|
|
while self.running:
|
|
|
|
try:
|
|
|
|
await asyncio.to_thread(
|
|
self.producer.send,
|
|
msg, properties
|
|
)
|
|
|
|
if self.metrics:
|
|
self.metrics.inc()
|
|
|
|
# Delivery success, break out of loop
|
|
break
|
|
|
|
except Exception as e:
|
|
print("Exception:", e, flush=True)
|
|
self.producer.close()
|
|
self.producer = None
|
|
|