mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-21 19:21:03 +02:00
74 lines
1.8 KiB
Python
74 lines
1.8 KiB
Python
|
|
from pulsar.schema import JsonSchema
|
|
import asyncio
|
|
|
|
class Producer:
|
|
|
|
def __init__(self, client, topic, schema, metrics=None,
|
|
chunking_enabled=True):
|
|
|
|
self.client = client
|
|
self.topic = topic
|
|
self.schema = schema
|
|
|
|
self.metrics = metrics
|
|
|
|
self.running = True
|
|
self.producer = None
|
|
|
|
self.chunking_enabled = chunking_enabled
|
|
|
|
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),
|
|
chunking_enabled = self.chunking_enabled,
|
|
)
|
|
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
|
|
|