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

Using subscriber.py

Browse source
This commit is contained in:
Cyber MacGeddon 2025-11-26 14:33:24 +00:00
parent a875d46e4d
commit ca9be84530
2 changed files with 14 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

18
trustgraph-base/trustgraph/base/subscriber.py
View file

@ -43,12 +43,18 @@ class Subscriber:
async def start(self): async def start(self):
self.consumer = self.client.subscribe( # Build subscribe arguments
topic = self.topic, subscribe_args = {
subscription_name = self.subscription, 'topic': self.topic,
consumer_name = self.consumer_name, 'subscription_name': self.subscription,
schema = JsonSchema(self.schema), 'consumer_name': self.consumer_name,
) }
# Only add schema if provided (omit if None)
if self.schema is not None:
subscribe_args['schema'] = JsonSchema(self.schema)
self.consumer = self.client.subscribe(**subscribe_args)
self.task = asyncio.create_task(self.run()) self.task = asyncio.create_task(self.run())

3
trustgraph-cli/trustgraph/cli/dump_queues.py
View file

@ -9,6 +9,7 @@ Uses TrustGraph's Subscriber abstraction for future-proof pub/sub compatibility.
""" """
import pulsar import pulsar
from pulsar.schema import BytesSchema
import sys import sys
import json import json
import asyncio import asyncio
@ -161,7 +162,7 @@ async def async_main(queues, output_file, pulsar_host, listener_name, subscriber
topic=queue_name, topic=queue_name,
subscription=subscriber_name, subscription=subscriber_name,
consumer_name=f"{subscriber_name}-{queue_name}", consumer_name=f"{subscriber_name}-{queue_name}",
schema=None, # Generic - no schema validation schema=None, # No schema - accept any message type
) )
await sub.start() await sub.start()
subscribers.append((queue_name, sub)) subscribers.append((queue_name, sub))